From f43228289b17ed01d2d231014e1a826b4507317e Mon Sep 17 00:00:00 2001 From: Karthik Kalyanaraman Date: Wed, 14 Oct 2026 23:30:55 -0700 Subject: [PATCH 1/4] Batch cancellation and steering hook registrations Signed-off-by: Karthik Kalyanaraman --- .changeset/quick-system-hook-batches.md | 7 + .../src/runtime/suspension-handler.test.ts | 184 ++++++++++++++++++ .../core/src/runtime/suspension-handler.ts | 108 +++++++++- packages/world-vercel/README.md | 16 ++ .../world-vercel/src/events-batch.test.ts | 134 ++++++++++++- packages/world-vercel/src/events-v4.ts | 50 ++++- packages/world-vercel/src/events.ts | 83 ++++++++ .../world-vercel/src/instrumentObject.test.ts | 25 +++ packages/world-vercel/src/instrumentObject.ts | 6 +- packages/world-vercel/src/storage.ts | 3 + .../src/trace-propagation.test.ts | 50 +++++ packages/world/README.md | 10 + packages/world/src/events.ts | 27 +++ packages/world/src/interfaces.ts | 18 ++ 14 files changed, 711 insertions(+), 10 deletions(-) create mode 100644 .changeset/quick-system-hook-batches.md diff --git a/.changeset/quick-system-hook-batches.md b/.changeset/quick-system-hook-batches.md new file mode 100644 index 0000000000..245972807b --- /dev/null +++ b/.changeset/quick-system-hook-batches.md @@ -0,0 +1,7 @@ +--- +"@workflow/core": minor +"@workflow/world": minor +"@workflow/world-vercel": minor +--- + +Coalesce generated cancellation and steering hook registrations into one request with one shared history catch-up, preserving individual hook commit and recovery semantics. diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index c82489cf6c..859b3eaa2e 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -3759,3 +3759,187 @@ describe('step-argument serialization failure', () => { expect(eventsCreate).not.toHaveBeenCalled(); }); }); + +describe('system hook request batching', () => { + const slotRun = { ...run, specVersion: 6 }; + const seed: Event = { + eventId: slotToEventId(5), + runId: run.runId, + eventType: 'run_started', + createdAt: new Date(), + } as Event; + function hooks(count = 2) { + return new Map( + Array.from({ length: count }, (_, i) => [ + `hook_${i}`, + { + type: 'hook' as const, + correlationId: `hook_${i}`, + token: `abrt_${i}`, + isSystem: true, + isWebhook: false, + }, + ]) + ); + } + function setup(count = 2) { + const create = vi.fn(async (_runId, event) => ({ event })); + const createHooks = vi.fn(async (_runId: string, entries: any[]) => { + const events = entries.map(({ event }, i) => ({ + ...event, + eventId: slotToEventId(6 + i), + runId: run.runId, + createdAt: new Date(), + })); + return { + results: events.map((event) => ({ status: 200, result: { event } })), + events, + cursor: `eid:${events.at(-1)?.eventId}`, + hasMore: false, + }; + }); + const world = createWorld(create); + world.events.createHooks = createHooks; + const eventLog = { events: [seed], cursor: `eid:${seed.eventId}` }; + const pending = hooks(count); + return { create, createHooks, world, eventLog, pending }; + } + afterEach(() => vi.unstubAllEnvs()); + + it('registers steer and cancel together and carries their complete suffix without a reload', async () => { + const t = setup(); + const result = await handleSuspension({ + ...t, + run: slotRun, + suspension: new WorkflowSuspension(t.pending, globalThis), + requestId: 'request', + }); + expect(t.createHooks).toHaveBeenCalledTimes(1); + expect( + t.createHooks.mock.calls[0][1].map((item) => item.event.correlationId) + ).toEqual(['hook_0', 'hook_1']); + expect(t.createHooks).toHaveBeenCalledWith(run.runId, expect.any(Array), { + requestId: 'request', + eventCount: 5, + }); + expect(t.create).not.toHaveBeenCalled(); + expect(t.eventLog.events.map((event) => event.eventId)).toEqual( + [5, 6, 7].map(slotToEventId) + ); + expect(result.eventLogCarriedForward).toBe(true); + }); + + it('preserves one-shot replay recovery telemetry on the shared request', async () => { + const t = setup(); + const reporter = new ReplayRecoveryReporter(2); + reporter.activate(); + await handleSuspension({ + ...t, + run: slotRun, + suspension: new WorkflowSuspension(t.pending, globalThis), + replayRecoveryReporter: reporter, + }); + expect(t.createHooks).toHaveBeenCalledWith( + run.runId, + expect.any(Array), + expect.objectContaining({ replayDivergenceCount: 2 }) + ); + }); + + it('drops a truncated shared page rather than introducing a hole', async () => { + const t = setup(); + const answer = t.createHooks.getMockImplementation(); + if (!answer) throw new Error('Missing batch mock'); + t.createHooks.mockImplementation(async (...args) => ({ + ...(await answer(...args)), + hasMore: true, + })); + const result = await handleSuspension({ + ...t, + run: slotRun, + suspension: new WorkflowSuspension(t.pending, globalThis), + }); + expect(t.eventLog.events).toEqual([seed]); + expect(result.eventLogCarriedForward).toBe(false); + }); + + it('preserves per-hook conflicts and requires a reload after a duplicate outcome', async () => { + const t = setup(); + t.createHooks.mockResolvedValue({ + results: [ + { status: 409, error: 'conflict', message: 'already exists' }, + { + status: 200, + result: { + event: { + ...seed, + eventType: 'hook_conflict', + correlationId: 'hook_1', + }, + }, + }, + ], + } as any); + const result = await handleSuspension({ + ...t, + run: slotRun, + suspension: new WorkflowSuspension(t.pending, globalThis), + }); + expect(result.hookConflictCorrelationIds).toEqual(['hook_1']); + expect(result.eventLogCarriedForward).toBe(false); + }); + + it.each([ + 'missing capability', + 'legacy run', + 'disabled', + ])('keeps the single path: %s', async (mode) => { + const t = setup(); + if (mode === 'missing capability') delete t.world.events.createHooks; + if (mode === 'disabled') vi.stubEnv('WORKFLOW_BATCH_TRANSITIONS', '0'); + await handleSuspension({ + ...t, + run: mode === 'legacy run' ? run : slotRun, + suspension: new WorkflowSuspension(t.pending, globalThis), + }); + expect(t.createHooks).not.toHaveBeenCalled(); + expect(t.create).toHaveBeenCalledTimes(2); + }); + + it.each([ + 'user hook', + 'force claim', + 'disposed hook', + 'metadata', + 'same token', + ])('preserves ordering for %s', async (mode) => { + const t = setup(); + const first = t.pending.get('hook_0')! as any; + if (mode === 'user hook') first.isSystem = false; + if (mode === 'force claim') first.force = true; + if (mode === 'disposed hook') first.disposed = true; + if (mode === 'metadata') first.metadata = { custom: true }; + if (mode === 'same token') t.pending.get('hook_1')!.token = first.token; + await handleSuspension({ + ...t, + run: slotRun, + suspension: new WorkflowSuspension(t.pending, globalThis), + }); + expect(t.createHooks).not.toHaveBeenCalled(); + expect(t.create).toHaveBeenCalledTimes(mode === 'disposed hook' ? 3 : 2); + }); + + it('bounds requests to 32 hooks and sends a final singleton through create', async () => { + const t = setup(65); + const result = await handleSuspension({ + ...t, + run: slotRun, + suspension: new WorkflowSuspension(t.pending, globalThis), + }); + expect(t.createHooks.mock.calls.map((call) => call[1].length)).toEqual([ + 32, 32, + ]); + expect(t.create).toHaveBeenCalledTimes(1); + expect(result.eventLogCarriedForward).toBe(false); + }); +}); diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index f3bcf5beb4..5edf5e8e7d 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -10,9 +10,11 @@ import { } from '@workflow/errors'; import { AttributeValidationError, + type BatchEventRequest, type CreateEventParams, type CreateEventRequest, type EventResult, + type HookBatchResult, type SerializedData, SPEC_VERSION_CURRENT, SPEC_VERSION_SUPPORTS_COMPRESSION, @@ -799,6 +801,109 @@ export async function handleSuspension({ ? eventLog.cursor : undefined; + // Generated abort hooks have independent tokens. User hooks, forced + // claims and same-token create/dispose sequences keep the existing path. + const createHooks = world.events.createHooks?.bind(world.events); + const batchableHooks = + isBatchTransitionsEnabled() && + createHooks !== undefined && + (run.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_SLOT_IDENTITY + ? hooksNeedingCreation.filter( + (item) => + item.isSystem === true && + item.token.startsWith('abrt_') && + !item.isWebhook && + !item.force && + !item.disposed && + !item.abortRequested && + item.metadata === undefined && + hookItemsByToken.get(item.token)?.length === 1 + ) + : []; + const hookBatchById = new Map Promise>(); + for (let offset = 0; offset < batchableHooks.length; offset += 32) { + const items = batchableHooks.slice(offset, offset + 32); + if (!createHooks || items.length < 2) continue; + let pending: Promise | undefined; + const start = () => { + if (!pending) { + eventWrites++; + const frontier = eventLog ? maxEventSlot(eventLog.events) : undefined; + const registrations: BatchEventRequest[] = items.map((item) => ({ + event: { + eventType: 'hook_created', + specVersion: SPEC_VERSION_CURRENT, + correlationId: item.correlationId, + eventData: { + token: item.token, + tokenRetentionUntil: item.tokenRetentionUntil, + isSystem: true, + isWebhook: false, + }, + }, + })); + pending = reporter + .withEventCreate( + { + requestId, + ...(frontier !== undefined ? { eventCount: frontier } : {}), + }, + (params) => createHooks(runId, registrations, params) + ) + .then((batch) => { + if (batch.results.length !== items.length) { + throw new WorkflowWorldError( + 'createHooks returned the wrong number of outcomes', + { code: 'SCHEMA_VALIDATION' } + ); + } + if ( + eventLog && + batch.events !== undefined && + batch.hasMore === false + ) { + reportedEvents += absorbSkippedSlotReport( + eventLog.events, + batch + ).added; + // Carry forward only when the shared suffix covers every + // outcome; any other write still invalidates it at return. + if ( + batch.results.every( + (item) => + item.error === undefined && + item.result.event && + eventLog.events.some( + (event) => event.eventId === item.result.event?.eventId + ) + ) + ) { + eventLog.cursor = batch.cursor ?? eventLog.cursor; + deltaAbsorbed = true; + } + } + return batch; + }); + } + return pending; + }; + for (const [index, item] of items.entries()) { + hookBatchById.set(item.correlationId, async () => { + const outcome = (await start()).results[index]; + if (outcome.error === undefined) return outcome.result; + if (outcome.status === 409 && outcome.error === 'conflict') { + throw new EntityConflictError(outcome.message); + } + if (outcome.status === 410) + throw new RunExpiredError(outcome.message, runId); + throw new WorkflowWorldError(outcome.message, { + status: outcome.status, + code: outcome.error, + }); + }); + } + } + const processHookGroup = async ( items: HookInvocationQueueItem[] ): Promise => { @@ -832,7 +937,8 @@ export async function handleSuspension({ queueItem, requestId, sinceCursor: hookDeltaCursor, - createEvent: createGuarded, + createEvent: + hookBatchById.get(queueItem.correlationId) ?? createGuarded, world, forceClaimVictimWakes, }); diff --git a/packages/world-vercel/README.md b/packages/world-vercel/README.md index ec80ca049e..740e370eff 100644 --- a/packages/world-vercel/README.md +++ b/packages/world-vercel/README.md @@ -38,6 +38,22 @@ It returns `undefined`, and writes stay on HTTP, when the transport is disabled or the World cannot hold a socket (a `projectConfig` World, as the CLI uses). +## System hook registration batches + +The runtime coalesces independent generated cancellation and steering hooks into +one `POST /v4/runs/:runId/events/hooks/batch` request (2–32 hooks). Each hook +keeps its existing commit and cross-region recovery protocol; this is not an +all-or-nothing transaction across the group. The response shares one history +suffix: consecutive new events need no history query, and intervening writes +require at most one shared catch-up query. Redis indexing remains awaited. + +Deploy the server route before releasing this adapter. The optional +`world.events.createHooks` method enables the optimization for slot-identity +runs; worlds without it, legacy runs, and `WORKFLOW_BATCH_TRANSITIONS=0` retain +individual writes. User hooks, metadata-bearing hooks, forced claims, and +same-token lifecycle operations retain their existing ordering. A failed batch +is not retried as individual writes while its outcome may still be unknown. + ## Stream writer sockets Stream writes upgrade to a WebSocket by default; set diff --git a/packages/world-vercel/src/events-batch.test.ts b/packages/world-vercel/src/events-batch.test.ts index 4b61a28b72..b91e36b1ed 100644 --- a/packages/world-vercel/src/events-batch.test.ts +++ b/packages/world-vercel/src/events-batch.test.ts @@ -3,7 +3,10 @@ import type { BatchEventRequest } from '@workflow/world'; import { decode, encode } from 'cbor-x'; import { MockAgent } from 'undici'; import { describe, expect, it } from 'vitest'; -import { createWorkflowRunEventBatch } from './events.js'; +import { + createWorkflowRunEventBatch, + createWorkflowRunHooks, +} from './events.js'; import { WORKFLOW_SERVER_URL_OVERRIDE } from './utils.js'; /** @@ -206,7 +209,8 @@ describe('createWorkflowRunEventBatch', () => { }); expect(requestBody).toBeDefined(); // biome-ignore lint/style/noNonNullAssertion: asserted above - const frames = decodeBatchFrames(requestBody!); + if (!requestBody) throw new Error('No batch request was sent'); + const frames = decodeBatchFrames(requestBody); expect(frames).toHaveLength(3); expect(frames.map((frame) => frame.meta.eventType)).toEqual([ 'step_completed', @@ -693,3 +697,129 @@ describe('createWorkflowRunEventBatch — retry-convergence and attribution', () ).rejects.toMatchObject({ code: 'SCHEMA_VALIDATION' }); }); }); + +describe('createWorkflowRunHooks', () => { + function hooks(): BatchEventRequest[] { + return [1, 2].map((n) => ({ + event: { + eventType: 'hook_created', + specVersion: 6, + correlationId: `hook_${n}`, + eventData: { token: `abrt_${n}`, isSystem: true, isWebhook: false }, + }, + })); + } + function body() { + const events = hooks().map(({ event }, i) => ({ + ...event, + eventId: slotEventId(6 + i), + runId: RUN_ID, + createdAt: CREATED_AT, + })); + return { + results: events.map((event) => ({ status: 200, event })), + events, + cursor: `eid:${slotEventId(7)}`, + hasMore: false, + }; + } + it('sends both registrations in one request with one shared frontier', async () => { + const agent = mockAgent(); + let requestBody: Uint8Array | undefined; + agent + .get(ORIGIN) + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/hooks/batch`, + method: 'POST', + body: (raw) => { + requestBody = new Uint8Array(Buffer.from(raw, 'binary')); + return true; + }, + }) + .reply(200, encode(body())); + const result = await createWorkflowRunHooks( + RUN_ID, + hooks(), + { eventCount: 5, requestId: 'test-request', replayDivergenceCount: 2 }, + { token: 'test', dispatcher: agent } + ); + const frames = decodeBatchFrames(requestBody!); + expect(frames).toHaveLength(2); + expect(frames.map((f) => f.meta.maxSlot)).toEqual([5, 5]); + expect(frames.map((f) => f.meta.vercelId)).toEqual([ + 'test-request', + 'test-request', + ]); + expect(frames.every((f) => f.payload.byteLength === 0)).toBe(true); + expect(result.results).toHaveLength(2); + expect(result.events?.map((e) => e.eventId)).toEqual([ + slotEventId(6), + slotEventId(7), + ]); + expect(result.events?.[0].createdAt).toBeInstanceOf(Date); + agent.assertNoPendingInterceptors(); + }); + it('preserves hook_conflict and duplicate outcomes', async () => { + const agent = mockAgent(); + const response = body(); + response.results[0].event = { + ...response.results[0].event, + eventType: 'hook_conflict', + eventData: { + token: 'abrt_1', + conflictingRunId: 'wrun_other', + }, + } as any; + response.results[1] = { + status: 409, + error: 'conflict', + message: 'already exists', + } as any; + agent + .get(ORIGIN) + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/hooks/batch`, + method: 'POST', + }) + .reply(200, encode(response)); + const result = await createWorkflowRunHooks(RUN_ID, hooks(), undefined, { + token: 'test', + dispatcher: agent, + }); + expect(result.results[0].result?.event?.eventType).toBe('hook_conflict'); + expect(result.results[1].error).toBe('conflict'); + }); + it('does not turn an ambiguous failure into individual writes', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/hooks/batch`, + method: 'POST', + }) + .reply(503, 'unavailable'); + await expect( + createWorkflowRunHooks(RUN_ID, hooks(), undefined, { + token: 'test', + dispatcher: agent, + }) + ).rejects.toThrow(); + agent.assertNoPendingInterceptors(); + }); + it('rejects malformed shared history', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/hooks/batch`, + method: 'POST', + }) + .reply(200, encode({ ...body(), events: [{ bad: true }] })); + await expect( + createWorkflowRunHooks(RUN_ID, hooks(), undefined, { + token: 'test', + dispatcher: agent, + }) + ).rejects.toThrow('invalid shared history'); + }); +}); diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index fe8c93e548..c470f2940a 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -1267,7 +1267,9 @@ export type CreateEventBatchV4Event = CreateEventV4InputBase & { export interface CreateEventBatchV4Input { runId: string; - /** Events in request order: the order they land in the run's log. */ + /** Hook-only request coalescing, with independent per-hook commits. */ + kind?: 'hooks'; + /** Request order. Hook-only registrations may commit in a different order. */ events: CreateEventBatchV4Event[]; } @@ -1284,6 +1286,9 @@ export type CreateEventBatchV4ItemResult = export interface CreateEventBatchV4Result { results: CreateEventBatchV4ItemResult[]; + events?: Event[]; + cursor?: string | null; + hasMore?: boolean; } const BatchItemFailureSchema = z.compile( @@ -1297,8 +1302,10 @@ const BatchItemFailureSchema = z.compile( /** * POST /api/v4/runs/:runId/events/batch * - * Appends an ordered batch of events to one run's log in a single durable - * write with per-event outcomes. The body is the events' single-POST frames + * The default kind appends an ordered transactional batch to one run's log. + * The hook-only kind coalesces independent registrations and shares their + * history catch-up; it does not make the group atomic. Both return per-event + * outcomes. The body is the events' single-POST frames * back-to-back (byte-identical framing, no batch-level meta); the response is * HTTP 200 CBOR `{ results }` whenever the batch was processed, one entry per * frame in request order. @@ -1341,7 +1348,8 @@ export async function createWorkflowRunEventsBatchV4( offset += frame.byteLength; } - const url = `${baseUrl}/v4/runs/${encodeURIComponent(input.runId)}/events/batch`; + const batchPath = input.kind === 'hooks' ? 'hooks/batch' : 'batch'; + const url = `${baseUrl}/v4/runs/${encodeURIComponent(input.runId)}/events/${batchPath}`; // Batch identity attributes (size, per-type shape) live on the // world.events.createBatch span (see instrumentObject); this transport // span carries only wire-level facts. workflow.event.type is deliberately @@ -1375,7 +1383,12 @@ export async function createWorkflowRunEventsBatchV4( const bodyBytes = new Uint8Array(await response.arrayBuffer()); const decoded = bodyBytes.byteLength > 0 - ? (decode(bodyBytes) as { results?: unknown[] }) + ? (decode(bodyBytes) as { + results?: unknown[]; + events?: unknown; + cursor?: unknown; + hasMore?: unknown; + }) : {}; // A 200 MUST carry exactly one outcome per submitted frame, in request // order: callers index `results` positionally. A missing / non-array / @@ -1425,7 +1438,13 @@ export async function createWorkflowRunEventsBatchV4( // Results are consumed positionally; an item whose committed event is // of a different type than the frame submitted at this index is a // server protocol violation, same as a wrong-length results array. - if (parsed.data.event.eventType !== eventType) { + if ( + parsed.data.event.eventType !== eventType && + !( + input.kind === 'hooks' && + parsed.data.event.eventType === 'hook_conflict' + ) + ) { throw new WorkflowWorldError( `v4 createEventBatch: result at index ${index} carries a ` + `${parsed.data.event.eventType} event, expected ${eventType}`, @@ -1436,6 +1455,25 @@ export async function createWorkflowRunEventsBatchV4( } ); + if (input.kind === 'hooks' && decoded.events !== undefined) { + const page = z + .object({ + events: z.array(VercelEventWireSchema), + cursor: z.string().nullable(), + hasMore: z.boolean(), + }) + .safeParse(decoded); + if (!page.success) { + throw new WorkflowWorldError( + 'v4 createHooks: invalid shared history page', + { + code: 'SCHEMA_VALIDATION', + cause: page.error, + } + ); + } + return { results, ...page.data }; + } return { results }; } diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index a6c0948bdd..c751419e54 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -43,6 +43,7 @@ import { type BatchEventRequest, type CreateEventBatchParams, type CreateEventParams, + type CreateHookBatchParams, type Event, type EventBatchResult, type EventDataPayloadField, @@ -51,6 +52,7 @@ import { entityResolveData, type GetEventParams, getEventDataPayloadField, + type HookBatchResult, isHookEventRequiringExistence, type ListEventsByCorrelationIdParams, type ListEventsParams, @@ -662,6 +664,87 @@ export async function createWorkflowRunEventBatch( }; } +/** Coalesce system-hook registrations; storage commits remain per hook. */ +export async function createWorkflowRunHooks( + runId: string, + events: BatchEventRequest[], + params?: CreateHookBatchParams, + config?: APIConfig +): Promise { + const tokens = new Set(); + const ids = new Set(); + if (events.length < 2 || events.length > 32) { + throw new WorkflowWorldError('createHooks requires 2–32 registrations', { + status: 400, + }); + } + for (const { event } of events) { + if ( + event.eventType !== 'hook_created' || + event.eventData.isSystem !== true || + !event.eventData.token.startsWith('abrt_') || + event.eventData.isWebhook || + event.eventData.force || + event.eventData.metadata !== undefined || + tokens.has(event.eventData.token) || + ids.has(event.correlationId) + ) { + throw new WorkflowWorldError( + 'createHooks requires distinct, metadata-free system hooks', + { status: 400 } + ); + } + tokens.add(event.eventData.token); + ids.add(event.correlationId); + } + // One attempt. An ambiguous transport failure redelivers the suspension; + // it never falls back to individual writes while siblings may still commit. + const wire = await createWorkflowRunEventsBatchV4( + { + runId, + kind: 'hooks', + events: events.map(({ event, occurredAt }, index) => { + const { payload, meta } = splitEventDataForV4(event); + return { + runId, + eventType: event.eventType, + specVersion: event.specVersion ?? 2, + correlationId: event.correlationId, + occurredAt: occurredAt ?? new Date(), + ...(params?.requestId ? { vercelId: params.requestId } : {}), + ...(index === 0 && params?.replayDivergenceCount !== undefined + ? { replayDivergenceCount: params.replayDivergenceCount } + : {}), + ...(params?.eventCount !== undefined + ? { maxSlot: params.eventCount } + : {}), + remoteRefBehavior: 'resolve' as const, + payload, + ...meta, + }; + }), + }, + config + ); + return { + results: wire.results.map((item) => + item.error !== undefined + ? item + : { + status: 200 as const, + result: item, + } + ), + ...(wire.events !== undefined + ? { + events: wire.events, + cursor: wire.cursor, + hasMore: wire.hasMore, + } + : {}), + }; +} + export async function createWorkflowRunEvent( id: string | null, data: T, diff --git a/packages/world-vercel/src/instrumentObject.test.ts b/packages/world-vercel/src/instrumentObject.test.ts index d7ac666989..96594601f6 100644 --- a/packages/world-vercel/src/instrumentObject.test.ts +++ b/packages/world-vercel/src/instrumentObject.test.ts @@ -84,4 +84,29 @@ describe('instrumentObject', () => { 'workflow.run.id': 'wrun_test', }); }); + it('tags system hook batches with their run and composition', async () => { + const events = instrumentObject('world.events', { + createHooks: vi.fn(async (_runId: string, _items: unknown[]) => ({ + results: [], + })), + }); + await events.createHooks('wrun_batch', [ + { event: { eventType: 'hook_created' } }, + { event: { eventType: 'hook_created' } }, + ]); + expect(mocks.trace).toHaveBeenCalledWith( + 'world.events.createHooks', + expect.anything(), + expect.any(Function) + ); + expect(mocks.span.setAttributes).toHaveBeenCalledWith( + expect.objectContaining({ 'workflow.run.id': 'wrun_batch' }) + ); + expect(mocks.span.setAttributes).toHaveBeenCalledWith( + expect.objectContaining({ + 'workflow.batch.size': 2, + 'workflow.batch.shape': 'hook_created:2', + }) + ); + }); }); diff --git a/packages/world-vercel/src/instrumentObject.ts b/packages/world-vercel/src/instrumentObject.ts index 0162d9071c..a1a239bbc4 100644 --- a/packages/world-vercel/src/instrumentObject.ts +++ b/packages/world-vercel/src/instrumentObject.ts @@ -25,6 +25,7 @@ const RUN_ID_ARG_INDEX_BY_METHOD: Record = { 'world.runs.experimentalSetAttributes': 0, 'world.steps.get': 0, 'world.events.create': 0, + 'world.events.createHooks': 0, 'world.events.get': 0, }; @@ -141,7 +142,10 @@ export function instrumentObject(prefix: string, o: T): T { // event. These live here (the world.events.createBatch span), not on // the transport span underneath. let batchAttributes: Record | undefined; - if (prefix === 'world.events' && methodName === 'createBatch') { + if ( + prefix === 'world.events' && + (methodName === 'createBatch' || methodName === 'createHooks') + ) { const events = args[1]; if (Array.isArray(events)) { const counts = new Map(); diff --git a/packages/world-vercel/src/storage.ts b/packages/world-vercel/src/storage.ts index 0c81d4d0d5..960a04e901 100644 --- a/packages/world-vercel/src/storage.ts +++ b/packages/world-vercel/src/storage.ts @@ -6,6 +6,7 @@ import type { import { createWorkflowRunEvent, createWorkflowRunEventBatch, + createWorkflowRunHooks, getEvent, getWorkflowRunEvents, } from './events.js'; @@ -58,6 +59,8 @@ export function createStorage(config?: APIConfig): Storage { ) => createWorkflowRunEvent(runId, data, params, config), createBatch: (runId, events, params) => createWorkflowRunEventBatch(runId, events, params, config), + createHooks: (runId, events, params) => + createWorkflowRunHooks(runId, events, params, config), get: (runId, eventId, params) => getEvent(runId, eventId, params, config), list: (params) => getWorkflowRunEvents(params, config), listByCorrelationId: (params) => getWorkflowRunEvents(params, config), diff --git a/packages/world-vercel/src/trace-propagation.test.ts b/packages/world-vercel/src/trace-propagation.test.ts index 6bd99a44e7..9457750abc 100644 --- a/packages/world-vercel/src/trace-propagation.test.ts +++ b/packages/world-vercel/src/trace-propagation.test.ts @@ -22,6 +22,7 @@ import { vi, } from 'vitest'; import { z } from 'zod'; +import { createWorkflowRunHooks } from './events.js'; import { createHookReceivedPreloadEventV4, getWorkflowRunEventsV4, @@ -610,3 +611,52 @@ describe.skipIf(WORKFLOW_SERVER_URL_OVERRIDE !== '')( }); } ); + +describe('system hook batch trace propagation', () => { + it('parents the shared registration HTTP request to the active workflow trace', async () => { + const fetchMock = vi.fn().mockResolvedValue( + cborResponse({ + results: [ + { status: 409, error: 'conflict', message: 'already registered' }, + { status: 409, error: 'conflict', message: 'already registered' }, + ], + }) + ); + vi.stubGlobal('fetch', fetchMock); + const tracer = otelTrace.getTracer('test'); + let traceId = ''; + let parentId = ''; + await tracer.startActiveSpan('flow-invocation', async (span) => { + traceId = span.spanContext().traceId; + parentId = span.spanContext().spanId; + await createWorkflowRunHooks( + 'wrun_trace', + ['cancel', 'steer'].map((name) => ({ + event: { + eventType: 'hook_created' as const, + specVersion: 6, + correlationId: `hook_${name}`, + eventData: { + token: `abrt_${name}`, + isSystem: true, + isWebhook: false, + }, + }, + })), + undefined, + { token: 'test-token' } + ); + span.end(); + }); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(String(fetchMock.mock.calls[0][0])).toContain('/events/hooks/batch'); + const headers = new Headers(fetchMock.mock.calls[0][1]?.headers); + const client = exporter + .getFinishedSpans() + .find((s) => s.name === 'http POST'); + expect(client?.parentSpanId).toBe(parentId); + expect(headers.get('traceparent')).toBe( + `00-${traceId}-${client?.spanContext().spanId}-01` + ); + }); +}); diff --git a/packages/world/README.md b/packages/world/README.md index 50d68b6ac3..e46973df46 100644 --- a/packages/world/README.md +++ b/packages/world/README.md @@ -6,6 +6,16 @@ This package defines the `World` interface that abstracts workflow storage, queu Used internally by `@workflow/core` and world implementations. Should not be used directly in application code. +## Optional system hook batches + +`events.createHooks(runId, registrations, params?)` coalesces 2–32 distinct, +metadata-free generated system hook registrations. Its per-item outcomes +represent independent commits, not a group transaction. Implementations must +settle all launched writes before reporting failure. When returning a shared +history suffix, use the caller's `eventCount` as the slot frontier and return +complete ordered history or accurate pagination. An incomplete suffix must +never be presented as complete. Omit the method to retain individual writes. + ## Stream writer transport lifetime A stateful `StreamWriteSession` may implement `release()` to retire an idle diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index 40ee6e0900..d37ce4d430 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -1262,6 +1262,33 @@ export type BatchEventItemResult = wait?: undefined; }; +/** Parameters for coalescing independent system-hook registrations. */ +export type CreateHookBatchParams = Pick< + CreateEventParams, + 'eventCount' | 'requestId' | 'replayDivergenceCount' +>; + +/** + * Hook registrations commit independently, with the same per-hook atomicity + * and retry semantics as `create`. This is request coalescing, not a transaction + * across hooks (hook storage and the run log may live in different regions). + */ +export interface HookBatchResult { + results: ( + | { + status: 200; + error?: undefined; + message?: undefined; + result: EventResult; + } + | { status: number; error: string; message: string; result?: undefined } + )[]; + /** One shared, dense suffix after the requested eventCount, when available. */ + events?: Event[]; + cursor?: string | null; + hasMore?: boolean; +} + /** Result of {@link Storage.events.createBatch}. */ export interface EventBatchResult { /** One entry per submitted event, in request order. */ diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts index cf8a419e78..dd469629aa 100644 --- a/packages/world/src/interfaces.ts +++ b/packages/world/src/interfaces.ts @@ -8,11 +8,13 @@ import type { CreateEventBatchParams, CreateEventParams, CreateEventRequest, + CreateHookBatchParams, Event, EventBatchResult, EventListResponse, EventResult, GetEventParams, + HookBatchResult, ListEventsByCorrelationIdParams, ListEventsParams, RunCreatedEventRequest, @@ -466,6 +468,22 @@ export interface Storage { params?: CreateEventBatchParams ): Promise; + /** + * Optional request coalescing for distinct, generated system hooks. Each + * registration uses `create`'s lifecycle protocol; the group is NOT atomic. + * All writes settle before returning or rejecting. A shared history suffix + * avoids each registration reading back its siblings' events. Worlds + * without this method keep using individual registrations. + * + * Only hook_created, isSystem:true, non-webhook, non-force, metadata-free + * registrations with distinct tokens and correlation IDs are accepted. + */ + createHooks?( + runId: string, + events: BatchEventRequest[], + params?: CreateHookBatchParams + ): Promise; + get( runId: string, eventId: string, From aa02d648f4ad46c27a5413cd6701e0e6d13b8689 Mon Sep 17 00:00:00 2001 From: Karthik Kalyanaraman Date: Wed, 14 Oct 2026 23:51:50 -0700 Subject: [PATCH 2/4] Negotiate hook batching before sending registrations Signed-off-by: Karthik Kalyanaraman --- packages/world-vercel/README.md | 6 +- packages/world-vercel/src/events-v4.ts | 16 +++- .../src/storage-hook-batch.test.ts | 96 +++++++++++++++++++ packages/world-vercel/src/storage.ts | 18 +++- packages/world-vercel/src/utils.ts | 6 ++ 5 files changed, 136 insertions(+), 6 deletions(-) create mode 100644 packages/world-vercel/src/storage-hook-batch.test.ts diff --git a/packages/world-vercel/README.md b/packages/world-vercel/README.md index 740e370eff..d76477d8f3 100644 --- a/packages/world-vercel/README.md +++ b/packages/world-vercel/README.md @@ -47,8 +47,10 @@ all-or-nothing transaction across the group. The response shares one history suffix: consecutive new events need no history query, and intervening writes require at most one shared catch-up query. Redis indexing remains awaited. -Deploy the server route before releasing this adapter. The optional -`world.events.createHooks` method enables the optimization for slot-identity +The adapter exposes the optional `world.events.createHooks` method only after +an existing replay response advertises `x-wf-hook-batch-version: 1`, with no +additional request. Older servers and unknown versions retain individual +writes, including after a rollback. The optimization applies to slot-identity runs; worlds without it, legacy runs, and `WORKFLOW_BATCH_TRANSITIONS=0` retain individual writes. User hooks, metadata-bearing hooks, forced claims, and same-token lifecycle operations retain their existing ordering. A failed batch diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index c470f2940a..9655a24a72 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -163,6 +163,15 @@ async function fetchV4( ), }); + // Replay setup already contacts the server before the runtime schedules + // hooks. Negotiate here without a probe or a speculative batch write. + // A missing/unknown version keeps older servers on individual writes. + if (response.headers.get('content-type')?.startsWith(V4_FRAME_CONTENT_TYPE)) { + config?.onHookBatchSupport?.( + response.headers.get('x-wf-hook-batch-version') === '1' + ); + } + if (!response.body) { noteEventsTransportOutcome(dispatcher); return response; @@ -1313,9 +1322,10 @@ const BatchItemFailureSchema = z.compile( * Slot-identity runs only: an older server 404s the route and a pre-slot * run is rejected with a 400. There is NO automatic fallback to single-event * posts on either: the runtime never sends a batch for a pre-slot run (it - * gates on the run's spec version), and against a backend without the route - * the batch fails and the suspension redelivers until the operator disables - * batching via `WORKFLOW_BATCH_TRANSITIONS=0`. Ambiguous failures (timeouts, + * gates on the run's spec version). For hook-only batches the storage adapter + * also requires support attested by the server's preceding replay response, + * so an older server keeps individual registrations without a failed POST. + * Ambiguous failures after a batch is sent (timeouts, * resets, 5xx, malformed responses) never convert to single posts either; * the wrapper either re-sends the SAME batch (only when its shape is * retry-convergent; see `createWorkflowRunEventBatch`) or surfaces the error diff --git a/packages/world-vercel/src/storage-hook-batch.test.ts b/packages/world-vercel/src/storage-hook-batch.test.ts new file mode 100644 index 0000000000..a32538c9ca --- /dev/null +++ b/packages/world-vercel/src/storage-hook-batch.test.ts @@ -0,0 +1,96 @@ +import { Buffer } from 'node:buffer'; +import { encode } from 'cbor-x'; +import { MockAgent } from 'undici'; +import { describe, expect, it } from 'vitest'; +import { encodeFrame, V4_FRAME_CONTENT_TYPE } from './frames.js'; +import { createStorage } from './storage.js'; +import { WORKFLOW_SERVER_URL_OVERRIDE } from './utils.js'; + +const ORIGIN = WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; + +function fixture() { + const agent = new MockAgent(); + agent.disableNetConnect(); + const storage = createStorage({ token: 'test', dispatcher: agent }); + const list = async (version?: string) => { + agent + .get(ORIGIN) + .intercept({ path: /\/api\/v4\/runs\/wrun_1\/events\?/, method: 'GET' }) + .reply( + 200, + Buffer.from(encodeFrame({ _end: 1, hasMore: false }, new Uint8Array())), + { + headers: { + 'content-type': V4_FRAME_CONTENT_TYPE, + ...(version === undefined + ? {} + : { 'x-wf-hook-batch-version': version }), + }, + } + ); + await storage.events.list({ runId: 'wrun_1' }); + }; + return { agent, storage, list }; +} + +describe('server-negotiated system hook batching', () => { + it('keeps individual registrations before setup and on older servers', async () => { + const { agent, storage, list } = fixture(); + try { + expect(storage.events.createHooks).toBeUndefined(); + await list(); + expect(storage.events.createHooks).toBeUndefined(); + agent.assertNoPendingInterceptors(); + } finally { + await agent.close(); + } + }); + + it('enables only the supported version and revokes it after a rollback', async () => { + const { agent, storage, list } = fixture(); + const other = createStorage(); + try { + await list('1'); + expect(storage.events.createHooks).toBeTypeOf('function'); + expect(other.events.createHooks).toBeUndefined(); + await list('2'); + expect(storage.events.createHooks).toBeUndefined(); + await list('1'); + expect(storage.events.createHooks).toBeTypeOf('function'); + await list(); + expect(storage.events.createHooks).toBeUndefined(); + agent.assertNoPendingInterceptors(); + } finally { + await agent.close(); + } + }); + + it('never falls back to individual writes after a batch is sent', async () => { + const { agent, storage, list } = fixture(); + try { + await list('1'); + agent + .get(ORIGIN) + .intercept({ + path: '/api/v4/runs/wrun_1/events/hooks/batch', + method: 'POST', + }) + .reply(404, encode({ error: 'not-found', message: 'Route missing' })); + await expect( + storage.events.createHooks?.( + 'wrun_1', + [1, 2].map((i) => ({ + event: { + eventType: 'hook_created' as const, + correlationId: `hook_${i}`, + eventData: { token: `abrt_${i}`, isSystem: true }, + }, + })) + ) + ).rejects.toThrow('Route missing'); + agent.assertNoPendingInterceptors(); + } finally { + await agent.close(); + } + }); +}); diff --git a/packages/world-vercel/src/storage.ts b/packages/world-vercel/src/storage.ts index 960a04e901..0ace9db5b8 100644 --- a/packages/world-vercel/src/storage.ts +++ b/packages/world-vercel/src/storage.ts @@ -25,6 +25,13 @@ import { getStep, listWorkflowRunSteps } from './steps.js'; import type { APIConfig } from './utils.js'; export function createStorage(config?: APIConfig): Storage { + let hookBatchSupported = false; + config = { + ...config, + onHookBatchSupport: (supported) => { + hookBatchSupported = supported; + }, + }; const snapshots = createSnapshotsStorage(config); const storage: Storage = { // Storage interface with namespaced methods @@ -74,11 +81,20 @@ export function createStorage(config?: APIConfig): Storage { }; // Instrument all storage methods with tracing + const events = instrumentObject('world.events', storage.events); + const createHooks = events.createHooks; + // Keep the capability dynamic: replay responses can enable it, while an + // older/rolled-back server disables it on the next setup response. Do not + // retry individual writes after an ambiguous batch failure. + Object.defineProperty(events, 'createHooks', { + enumerable: true, + get: () => (hookBatchSupported ? createHooks : undefined), + }); // NOTE: Span names are lowercase per OTEL semantic conventions return { runs: instrumentObject('world.runs', storage.runs), steps: instrumentObject('world.steps', storage.steps), - events: instrumentObject('world.events', storage.events), + events, hooks: instrumentObject('world.hooks', storage.hooks), experimental_snapshots: instrumentObject( 'world.experimental_snapshots', diff --git a/packages/world-vercel/src/utils.ts b/packages/world-vercel/src/utils.ts index 3cba7d0808..ff81666683 100644 --- a/packages/world-vercel/src/utils.ts +++ b/packages/world-vercel/src/utils.ts @@ -140,6 +140,12 @@ export interface APIConfig { * @internal */ mintedSpecVersion?: number; + /** + * Observe server support on replay responses, before scheduling writes. + * Scoped to one storage instance; absent headers disable hook batching. + * @internal + */ + onHookBatchSupport?: (supported: boolean) => void; } export const DEFAULT_RESOLVE_DATA_OPTION = 'all'; From 249d8df210892c72410c4d6d7676d16a68bc472d Mon Sep 17 00:00:00 2001 From: Karthik Kalyanaraman Date: Thu, 15 Oct 2026 00:18:54 -0700 Subject: [PATCH 3/4] Reconcile concurrent hook registrations through the existing event API Signed-off-by: Karthik Kalyanaraman --- .changeset/quick-system-hook-batches.md | 7 - .changeset/shared-hook-history.md | 7 + .../docs/v5/how-it-works/event-sourcing.mdx | 2 +- packages/core/src/runtime.test.ts | 117 ++++++ .../concurrent-hook-registration.test.ts | 387 ++++++++++++++++++ packages/core/src/runtime/helpers.ts | 2 +- .../src/runtime/suspension-handler.test.ts | 184 --------- .../core/src/runtime/suspension-handler.ts | 207 +++++----- packages/world-vercel/README.md | 26 +- .../world-vercel/src/events-batch.test.ts | 134 +----- packages/world-vercel/src/events-v4.ts | 66 +-- packages/world-vercel/src/events.test.ts | 105 +++++ packages/world-vercel/src/events.ts | 104 +---- .../world-vercel/src/instrumentObject.test.ts | 25 -- packages/world-vercel/src/instrumentObject.ts | 6 +- .../src/storage-hook-batch.test.ts | 96 ----- packages/world-vercel/src/storage.ts | 21 +- .../src/trace-propagation.test.ts | 50 --- packages/world-vercel/src/utils.ts | 6 - packages/world/README.md | 20 +- packages/world/src/events.ts | 35 +- packages/world/src/interfaces.ts | 18 - 22 files changed, 769 insertions(+), 856 deletions(-) delete mode 100644 .changeset/quick-system-hook-batches.md create mode 100644 .changeset/shared-hook-history.md create mode 100644 packages/core/src/runtime/concurrent-hook-registration.test.ts delete mode 100644 packages/world-vercel/src/storage-hook-batch.test.ts diff --git a/.changeset/quick-system-hook-batches.md b/.changeset/quick-system-hook-batches.md deleted file mode 100644 index 245972807b..0000000000 --- a/.changeset/quick-system-hook-batches.md +++ /dev/null @@ -1,7 +0,0 @@ ---- -"@workflow/core": minor -"@workflow/world": minor -"@workflow/world-vercel": minor ---- - -Coalesce generated cancellation and steering hook registrations into one request with one shared history catch-up, preserving individual hook commit and recovery semantics. 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..0cf0257e85 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 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/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..b692eec6e9 --- /dev/null +++ b/packages/core/src/runtime/concurrent-hook-registration.test.ts @@ -0,0 +1,387 @@ +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 invoke = ( + options: Partial[0]> = {} + ) => + handleSuspension({ + world, + run, + eventLog, + requestId: 'request', + suspension: new WorkflowSuspension(pending, globalThis), + ...options, + }); + return { create, answer, world, 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(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', + 'gap between', + 'missing event', + 'missing cursor', + 'duplicate slot', + 'old event', + 'wrong run', + 'wrong correlation', + 'wrong token', + 'wrong type', + 'non-slot ID', + 'truncated page', + 'lazy metadata', + 'omitted metadata', + ])('retains the reload path without changing the log: %s', async (mode) => { + 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); + }); + + 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); + }); + + 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); + }); + + 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); + expect(t.eventLog.events).toEqual(t.seed); + }); + + 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..349f1f7c1c 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 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.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index 859b3eaa2e..c82489cf6c 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -3759,187 +3759,3 @@ describe('step-argument serialization failure', () => { expect(eventsCreate).not.toHaveBeenCalled(); }); }); - -describe('system hook request batching', () => { - const slotRun = { ...run, specVersion: 6 }; - const seed: Event = { - eventId: slotToEventId(5), - runId: run.runId, - eventType: 'run_started', - createdAt: new Date(), - } as Event; - function hooks(count = 2) { - return new Map( - Array.from({ length: count }, (_, i) => [ - `hook_${i}`, - { - type: 'hook' as const, - correlationId: `hook_${i}`, - token: `abrt_${i}`, - isSystem: true, - isWebhook: false, - }, - ]) - ); - } - function setup(count = 2) { - const create = vi.fn(async (_runId, event) => ({ event })); - const createHooks = vi.fn(async (_runId: string, entries: any[]) => { - const events = entries.map(({ event }, i) => ({ - ...event, - eventId: slotToEventId(6 + i), - runId: run.runId, - createdAt: new Date(), - })); - return { - results: events.map((event) => ({ status: 200, result: { event } })), - events, - cursor: `eid:${events.at(-1)?.eventId}`, - hasMore: false, - }; - }); - const world = createWorld(create); - world.events.createHooks = createHooks; - const eventLog = { events: [seed], cursor: `eid:${seed.eventId}` }; - const pending = hooks(count); - return { create, createHooks, world, eventLog, pending }; - } - afterEach(() => vi.unstubAllEnvs()); - - it('registers steer and cancel together and carries their complete suffix without a reload', async () => { - const t = setup(); - const result = await handleSuspension({ - ...t, - run: slotRun, - suspension: new WorkflowSuspension(t.pending, globalThis), - requestId: 'request', - }); - expect(t.createHooks).toHaveBeenCalledTimes(1); - expect( - t.createHooks.mock.calls[0][1].map((item) => item.event.correlationId) - ).toEqual(['hook_0', 'hook_1']); - expect(t.createHooks).toHaveBeenCalledWith(run.runId, expect.any(Array), { - requestId: 'request', - eventCount: 5, - }); - expect(t.create).not.toHaveBeenCalled(); - expect(t.eventLog.events.map((event) => event.eventId)).toEqual( - [5, 6, 7].map(slotToEventId) - ); - expect(result.eventLogCarriedForward).toBe(true); - }); - - it('preserves one-shot replay recovery telemetry on the shared request', async () => { - const t = setup(); - const reporter = new ReplayRecoveryReporter(2); - reporter.activate(); - await handleSuspension({ - ...t, - run: slotRun, - suspension: new WorkflowSuspension(t.pending, globalThis), - replayRecoveryReporter: reporter, - }); - expect(t.createHooks).toHaveBeenCalledWith( - run.runId, - expect.any(Array), - expect.objectContaining({ replayDivergenceCount: 2 }) - ); - }); - - it('drops a truncated shared page rather than introducing a hole', async () => { - const t = setup(); - const answer = t.createHooks.getMockImplementation(); - if (!answer) throw new Error('Missing batch mock'); - t.createHooks.mockImplementation(async (...args) => ({ - ...(await answer(...args)), - hasMore: true, - })); - const result = await handleSuspension({ - ...t, - run: slotRun, - suspension: new WorkflowSuspension(t.pending, globalThis), - }); - expect(t.eventLog.events).toEqual([seed]); - expect(result.eventLogCarriedForward).toBe(false); - }); - - it('preserves per-hook conflicts and requires a reload after a duplicate outcome', async () => { - const t = setup(); - t.createHooks.mockResolvedValue({ - results: [ - { status: 409, error: 'conflict', message: 'already exists' }, - { - status: 200, - result: { - event: { - ...seed, - eventType: 'hook_conflict', - correlationId: 'hook_1', - }, - }, - }, - ], - } as any); - const result = await handleSuspension({ - ...t, - run: slotRun, - suspension: new WorkflowSuspension(t.pending, globalThis), - }); - expect(result.hookConflictCorrelationIds).toEqual(['hook_1']); - expect(result.eventLogCarriedForward).toBe(false); - }); - - it.each([ - 'missing capability', - 'legacy run', - 'disabled', - ])('keeps the single path: %s', async (mode) => { - const t = setup(); - if (mode === 'missing capability') delete t.world.events.createHooks; - if (mode === 'disabled') vi.stubEnv('WORKFLOW_BATCH_TRANSITIONS', '0'); - await handleSuspension({ - ...t, - run: mode === 'legacy run' ? run : slotRun, - suspension: new WorkflowSuspension(t.pending, globalThis), - }); - expect(t.createHooks).not.toHaveBeenCalled(); - expect(t.create).toHaveBeenCalledTimes(2); - }); - - it.each([ - 'user hook', - 'force claim', - 'disposed hook', - 'metadata', - 'same token', - ])('preserves ordering for %s', async (mode) => { - const t = setup(); - const first = t.pending.get('hook_0')! as any; - if (mode === 'user hook') first.isSystem = false; - if (mode === 'force claim') first.force = true; - if (mode === 'disposed hook') first.disposed = true; - if (mode === 'metadata') first.metadata = { custom: true }; - if (mode === 'same token') t.pending.get('hook_1')!.token = first.token; - await handleSuspension({ - ...t, - run: slotRun, - suspension: new WorkflowSuspension(t.pending, globalThis), - }); - expect(t.createHooks).not.toHaveBeenCalled(); - expect(t.create).toHaveBeenCalledTimes(mode === 'disposed hook' ? 3 : 2); - }); - - it('bounds requests to 32 hooks and sends a final singleton through create', async () => { - const t = setup(65); - const result = await handleSuspension({ - ...t, - run: slotRun, - suspension: new WorkflowSuspension(t.pending, globalThis), - }); - expect(t.createHooks.mock.calls.map((call) => call[1].length)).toEqual([ - 32, 32, - ]); - expect(t.create).toHaveBeenCalledTimes(1); - expect(result.eventLogCarriedForward).toBe(false); - }); -}); diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index 5edf5e8e7d..75033aeb03 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -10,11 +10,11 @@ import { } from '@workflow/errors'; import { AttributeValidationError, - type BatchEventRequest, type CreateEventParams, type CreateEventRequest, + type Event, type EventResult, - type HookBatchResult, + eventIdToSlot, type SerializedData, SPEC_VERSION_CURRENT, SPEC_VERSION_SUPPORTS_COMPRESSION, @@ -54,6 +54,7 @@ import { import { absorbSkippedSlotReport, type EventCreator, + findEventSlotGap, type LoadedEventLog, maxEventSlot, mergeReportedEvents, @@ -77,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 @@ -310,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 @@ -801,108 +803,39 @@ export async function handleSuspension({ ? eventLog.cursor : undefined; - // Generated abort hooks have independent tokens. User hooks, forced - // claims and same-token create/dispose sequences keep the existing path. - const createHooks = world.events.createHooks?.bind(world.events); - const batchableHooks = - isBatchTransitionsEnabled() && - createHooks !== undefined && - (run.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_SLOT_IDENTITY + // 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. + const concurrentHooks = + eventLog && (run.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_SLOT_IDENTITY ? hooksNeedingCreation.filter( (item) => - item.isSystem === true && - item.token.startsWith('abrt_') && - !item.isWebhook && !item.force && !item.disposed && !item.abortRequested && - item.metadata === undefined && hookItemsByToken.get(item.token)?.length === 1 ) : []; - const hookBatchById = new Map Promise>(); - for (let offset = 0; offset < batchableHooks.length; offset += 32) { - const items = batchableHooks.slice(offset, offset + 32); - if (!createHooks || items.length < 2) continue; - let pending: Promise | undefined; - const start = () => { - if (!pending) { - eventWrites++; - const frontier = eventLog ? maxEventSlot(eventLog.events) : undefined; - const registrations: BatchEventRequest[] = items.map((item) => ({ - event: { - eventType: 'hook_created', - specVersion: SPEC_VERSION_CURRENT, - correlationId: item.correlationId, - eventData: { - token: item.token, - tokenRetentionUntil: item.tokenRetentionUntil, - isSystem: true, - isWebhook: false, - }, - }, - })); - pending = reporter - .withEventCreate( - { - requestId, - ...(frontier !== undefined ? { eventCount: frontier } : {}), - }, - (params) => createHooks(runId, registrations, params) - ) - .then((batch) => { - if (batch.results.length !== items.length) { - throw new WorkflowWorldError( - 'createHooks returned the wrong number of outcomes', - { code: 'SCHEMA_VALIDATION' } - ); - } - if ( - eventLog && - batch.events !== undefined && - batch.hasMore === false - ) { - reportedEvents += absorbSkippedSlotReport( - eventLog.events, - batch - ).added; - // Carry forward only when the shared suffix covers every - // outcome; any other write still invalidates it at return. - if ( - batch.results.every( - (item) => - item.error === undefined && - item.result.event && - eventLog.events.some( - (event) => event.eventId === item.result.event?.eventId - ) - ) - ) { - eventLog.cursor = batch.cursor ?? eventLog.cursor; - deltaAbsorbed = true; - } - } - return batch; - }); - } - return pending; - }; - for (const [index, item] of items.entries()) { - hookBatchById.set(item.correlationId, async () => { - const outcome = (await start()).results[index]; - if (outcome.error === undefined) return outcome.result; - if (outcome.status === 409 && outcome.error === 'conflict') { - throw new EntityConflictError(outcome.message); - } - if (outcome.status === 410) - throw new RunExpiredError(outcome.message, runId); - throw new WorkflowWorldError(outcome.message, { - status: outcome.status, - code: outcome.error, - }); - }); - } - } + 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[] @@ -937,8 +870,9 @@ export async function handleSuspension({ queueItem, requestId, sinceCursor: hookDeltaCursor, - createEvent: - hookBatchById.get(queueItem.correlationId) ?? createGuarded, + createEvent: concurrentHookIds.has(queueItem.correlationId) + ? createConcurrentHook + : createGuarded, world, forceClaimVictimWakes, }); @@ -2059,6 +1993,59 @@ 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. + let concurrentHooksCarriedForward = false; + if ( + eventLog && + hookFrontier !== undefined && + concurrentHookIds.size > 1 && + hookResults.size === concurrentHookIds.size && + eventWrites === hookResults.size && + maxEventSlot(eventLog.events) === hookFrontier && + findEventSlotGap(eventLog.events) === undefined + ) { + const committed: { event: Event; cursor?: string }[] = []; + for (const item of concurrentHooks) { + const result = hookResults.get(item.correlationId)!; + const event = result.event; + if ( + !event || + result.hasMore === true || + event.runId !== runId || + event.correlationId !== item.correlationId || + (event.eventType !== 'hook_created' && + event.eventType !== 'hook_conflict') || + event.eventData.token !== item.token || + (event.eventType === 'hook_created' && + ((event.eventData.metadata !== undefined && + !(event.eventData.metadata instanceof Uint8Array)) || + (item.metadata !== undefined && + event.eventData.metadata === undefined))) || + eventIdToSlot(event.eventId) === null + ) { + break; + } + committed.push({ event, cursor: result.eventCursor }); + } + committed.sort((a, b) => a.event.eventId.localeCompare(b.event.eventId)); + const last = committed.at(-1); + if ( + committed.length === concurrentHookIds.size && + last?.cursor && + committed.every( + ({ event }, index) => + eventIdToSlot(event.eventId) === hookFrontier + index + 1 + ) + ) { + eventLog.events.push(...committed.map(({ event }) => event)); + eventLog.cursor = last.cursor; + concurrentHooksCarriedForward = true; + } + } + // 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. @@ -2126,10 +2113,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 d76477d8f3..2f713f0dd2 100644 --- a/packages/world-vercel/README.md +++ b/packages/world-vercel/README.md @@ -1,5 +1,13 @@ # @workflow/world-vercel +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. + Production workflow backend for Vercel platform deployments. Integrates with Vercel's infrastructure for storage, queuing, and authentication. Handles workflow persistence and scaling in production environments. @@ -38,24 +46,6 @@ It returns `undefined`, and writes stay on HTTP, when the transport is disabled or the World cannot hold a socket (a `projectConfig` World, as the CLI uses). -## System hook registration batches - -The runtime coalesces independent generated cancellation and steering hooks into -one `POST /v4/runs/:runId/events/hooks/batch` request (2–32 hooks). Each hook -keeps its existing commit and cross-region recovery protocol; this is not an -all-or-nothing transaction across the group. The response shares one history -suffix: consecutive new events need no history query, and intervening writes -require at most one shared catch-up query. Redis indexing remains awaited. - -The adapter exposes the optional `world.events.createHooks` method only after -an existing replay response advertises `x-wf-hook-batch-version: 1`, with no -additional request. Older servers and unknown versions retain individual -writes, including after a rollback. The optimization applies to slot-identity -runs; worlds without it, legacy runs, and `WORKFLOW_BATCH_TRANSITIONS=0` retain -individual writes. User hooks, metadata-bearing hooks, forced claims, and -same-token lifecycle operations retain their existing ordering. A failed batch -is not retried as individual writes while its outcome may still be unknown. - ## Stream writer sockets Stream writes upgrade to a WebSocket by default; set diff --git a/packages/world-vercel/src/events-batch.test.ts b/packages/world-vercel/src/events-batch.test.ts index b91e36b1ed..4b61a28b72 100644 --- a/packages/world-vercel/src/events-batch.test.ts +++ b/packages/world-vercel/src/events-batch.test.ts @@ -3,10 +3,7 @@ import type { BatchEventRequest } from '@workflow/world'; import { decode, encode } from 'cbor-x'; import { MockAgent } from 'undici'; import { describe, expect, it } from 'vitest'; -import { - createWorkflowRunEventBatch, - createWorkflowRunHooks, -} from './events.js'; +import { createWorkflowRunEventBatch } from './events.js'; import { WORKFLOW_SERVER_URL_OVERRIDE } from './utils.js'; /** @@ -209,8 +206,7 @@ describe('createWorkflowRunEventBatch', () => { }); expect(requestBody).toBeDefined(); // biome-ignore lint/style/noNonNullAssertion: asserted above - if (!requestBody) throw new Error('No batch request was sent'); - const frames = decodeBatchFrames(requestBody); + const frames = decodeBatchFrames(requestBody!); expect(frames).toHaveLength(3); expect(frames.map((frame) => frame.meta.eventType)).toEqual([ 'step_completed', @@ -697,129 +693,3 @@ describe('createWorkflowRunEventBatch — retry-convergence and attribution', () ).rejects.toMatchObject({ code: 'SCHEMA_VALIDATION' }); }); }); - -describe('createWorkflowRunHooks', () => { - function hooks(): BatchEventRequest[] { - return [1, 2].map((n) => ({ - event: { - eventType: 'hook_created', - specVersion: 6, - correlationId: `hook_${n}`, - eventData: { token: `abrt_${n}`, isSystem: true, isWebhook: false }, - }, - })); - } - function body() { - const events = hooks().map(({ event }, i) => ({ - ...event, - eventId: slotEventId(6 + i), - runId: RUN_ID, - createdAt: CREATED_AT, - })); - return { - results: events.map((event) => ({ status: 200, event })), - events, - cursor: `eid:${slotEventId(7)}`, - hasMore: false, - }; - } - it('sends both registrations in one request with one shared frontier', async () => { - const agent = mockAgent(); - let requestBody: Uint8Array | undefined; - agent - .get(ORIGIN) - .intercept({ - path: `/api/v4/runs/${RUN_ID}/events/hooks/batch`, - method: 'POST', - body: (raw) => { - requestBody = new Uint8Array(Buffer.from(raw, 'binary')); - return true; - }, - }) - .reply(200, encode(body())); - const result = await createWorkflowRunHooks( - RUN_ID, - hooks(), - { eventCount: 5, requestId: 'test-request', replayDivergenceCount: 2 }, - { token: 'test', dispatcher: agent } - ); - const frames = decodeBatchFrames(requestBody!); - expect(frames).toHaveLength(2); - expect(frames.map((f) => f.meta.maxSlot)).toEqual([5, 5]); - expect(frames.map((f) => f.meta.vercelId)).toEqual([ - 'test-request', - 'test-request', - ]); - expect(frames.every((f) => f.payload.byteLength === 0)).toBe(true); - expect(result.results).toHaveLength(2); - expect(result.events?.map((e) => e.eventId)).toEqual([ - slotEventId(6), - slotEventId(7), - ]); - expect(result.events?.[0].createdAt).toBeInstanceOf(Date); - agent.assertNoPendingInterceptors(); - }); - it('preserves hook_conflict and duplicate outcomes', async () => { - const agent = mockAgent(); - const response = body(); - response.results[0].event = { - ...response.results[0].event, - eventType: 'hook_conflict', - eventData: { - token: 'abrt_1', - conflictingRunId: 'wrun_other', - }, - } as any; - response.results[1] = { - status: 409, - error: 'conflict', - message: 'already exists', - } as any; - agent - .get(ORIGIN) - .intercept({ - path: `/api/v4/runs/${RUN_ID}/events/hooks/batch`, - method: 'POST', - }) - .reply(200, encode(response)); - const result = await createWorkflowRunHooks(RUN_ID, hooks(), undefined, { - token: 'test', - dispatcher: agent, - }); - expect(result.results[0].result?.event?.eventType).toBe('hook_conflict'); - expect(result.results[1].error).toBe('conflict'); - }); - it('does not turn an ambiguous failure into individual writes', async () => { - const agent = mockAgent(); - agent - .get(ORIGIN) - .intercept({ - path: `/api/v4/runs/${RUN_ID}/events/hooks/batch`, - method: 'POST', - }) - .reply(503, 'unavailable'); - await expect( - createWorkflowRunHooks(RUN_ID, hooks(), undefined, { - token: 'test', - dispatcher: agent, - }) - ).rejects.toThrow(); - agent.assertNoPendingInterceptors(); - }); - it('rejects malformed shared history', async () => { - const agent = mockAgent(); - agent - .get(ORIGIN) - .intercept({ - path: `/api/v4/runs/${RUN_ID}/events/hooks/batch`, - method: 'POST', - }) - .reply(200, encode({ ...body(), events: [{ bad: true }] })); - await expect( - createWorkflowRunHooks(RUN_ID, hooks(), undefined, { - token: 'test', - dispatcher: agent, - }) - ).rejects.toThrow('invalid shared history'); - }); -}); diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index 9655a24a72..fe8c93e548 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -163,15 +163,6 @@ async function fetchV4( ), }); - // Replay setup already contacts the server before the runtime schedules - // hooks. Negotiate here without a probe or a speculative batch write. - // A missing/unknown version keeps older servers on individual writes. - if (response.headers.get('content-type')?.startsWith(V4_FRAME_CONTENT_TYPE)) { - config?.onHookBatchSupport?.( - response.headers.get('x-wf-hook-batch-version') === '1' - ); - } - if (!response.body) { noteEventsTransportOutcome(dispatcher); return response; @@ -1276,9 +1267,7 @@ export type CreateEventBatchV4Event = CreateEventV4InputBase & { export interface CreateEventBatchV4Input { runId: string; - /** Hook-only request coalescing, with independent per-hook commits. */ - kind?: 'hooks'; - /** Request order. Hook-only registrations may commit in a different order. */ + /** Events in request order: the order they land in the run's log. */ events: CreateEventBatchV4Event[]; } @@ -1295,9 +1284,6 @@ export type CreateEventBatchV4ItemResult = export interface CreateEventBatchV4Result { results: CreateEventBatchV4ItemResult[]; - events?: Event[]; - cursor?: string | null; - hasMore?: boolean; } const BatchItemFailureSchema = z.compile( @@ -1311,10 +1297,8 @@ const BatchItemFailureSchema = z.compile( /** * POST /api/v4/runs/:runId/events/batch * - * The default kind appends an ordered transactional batch to one run's log. - * The hook-only kind coalesces independent registrations and shares their - * history catch-up; it does not make the group atomic. Both return per-event - * outcomes. The body is the events' single-POST frames + * Appends an ordered batch of events to one run's log in a single durable + * write with per-event outcomes. The body is the events' single-POST frames * back-to-back (byte-identical framing, no batch-level meta); the response is * HTTP 200 CBOR `{ results }` whenever the batch was processed, one entry per * frame in request order. @@ -1322,10 +1306,9 @@ const BatchItemFailureSchema = z.compile( * Slot-identity runs only: an older server 404s the route and a pre-slot * run is rejected with a 400. There is NO automatic fallback to single-event * posts on either: the runtime never sends a batch for a pre-slot run (it - * gates on the run's spec version). For hook-only batches the storage adapter - * also requires support attested by the server's preceding replay response, - * so an older server keeps individual registrations without a failed POST. - * Ambiguous failures after a batch is sent (timeouts, + * gates on the run's spec version), and against a backend without the route + * the batch fails and the suspension redelivers until the operator disables + * batching via `WORKFLOW_BATCH_TRANSITIONS=0`. Ambiguous failures (timeouts, * resets, 5xx, malformed responses) never convert to single posts either; * the wrapper either re-sends the SAME batch (only when its shape is * retry-convergent; see `createWorkflowRunEventBatch`) or surfaces the error @@ -1358,8 +1341,7 @@ export async function createWorkflowRunEventsBatchV4( offset += frame.byteLength; } - const batchPath = input.kind === 'hooks' ? 'hooks/batch' : 'batch'; - const url = `${baseUrl}/v4/runs/${encodeURIComponent(input.runId)}/events/${batchPath}`; + const url = `${baseUrl}/v4/runs/${encodeURIComponent(input.runId)}/events/batch`; // Batch identity attributes (size, per-type shape) live on the // world.events.createBatch span (see instrumentObject); this transport // span carries only wire-level facts. workflow.event.type is deliberately @@ -1393,12 +1375,7 @@ export async function createWorkflowRunEventsBatchV4( const bodyBytes = new Uint8Array(await response.arrayBuffer()); const decoded = bodyBytes.byteLength > 0 - ? (decode(bodyBytes) as { - results?: unknown[]; - events?: unknown; - cursor?: unknown; - hasMore?: unknown; - }) + ? (decode(bodyBytes) as { results?: unknown[] }) : {}; // A 200 MUST carry exactly one outcome per submitted frame, in request // order: callers index `results` positionally. A missing / non-array / @@ -1448,13 +1425,7 @@ export async function createWorkflowRunEventsBatchV4( // Results are consumed positionally; an item whose committed event is // of a different type than the frame submitted at this index is a // server protocol violation, same as a wrong-length results array. - if ( - parsed.data.event.eventType !== eventType && - !( - input.kind === 'hooks' && - parsed.data.event.eventType === 'hook_conflict' - ) - ) { + if (parsed.data.event.eventType !== eventType) { throw new WorkflowWorldError( `v4 createEventBatch: result at index ${index} carries a ` + `${parsed.data.event.eventType} event, expected ${eventType}`, @@ -1465,25 +1436,6 @@ export async function createWorkflowRunEventsBatchV4( } ); - if (input.kind === 'hooks' && decoded.events !== undefined) { - const page = z - .object({ - events: z.array(VercelEventWireSchema), - cursor: z.string().nullable(), - hasMore: z.boolean(), - }) - .safeParse(decoded); - if (!page.success) { - throw new WorkflowWorldError( - 'v4 createHooks: invalid shared history page', - { - code: 'SCHEMA_VALIDATION', - cause: page.error, - } - ); - } - return { results, ...page.data }; - } return { results }; } diff --git a/packages/world-vercel/src/events.test.ts b/packages/world-vercel/src/events.test.ts index 56199ef9b3..fa39146e50 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,110 @@ 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('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 c751419e54..659702cf4c 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -43,7 +43,6 @@ import { type BatchEventRequest, type CreateEventBatchParams, type CreateEventParams, - type CreateHookBatchParams, type Event, type EventBatchResult, type EventDataPayloadField, @@ -52,7 +51,6 @@ import { entityResolveData, type GetEventParams, getEventDataPayloadField, - type HookBatchResult, isHookEventRequiringExistence, type ListEventsByCorrelationIdParams, type ListEventsParams, @@ -664,87 +662,6 @@ export async function createWorkflowRunEventBatch( }; } -/** Coalesce system-hook registrations; storage commits remain per hook. */ -export async function createWorkflowRunHooks( - runId: string, - events: BatchEventRequest[], - params?: CreateHookBatchParams, - config?: APIConfig -): Promise { - const tokens = new Set(); - const ids = new Set(); - if (events.length < 2 || events.length > 32) { - throw new WorkflowWorldError('createHooks requires 2–32 registrations', { - status: 400, - }); - } - for (const { event } of events) { - if ( - event.eventType !== 'hook_created' || - event.eventData.isSystem !== true || - !event.eventData.token.startsWith('abrt_') || - event.eventData.isWebhook || - event.eventData.force || - event.eventData.metadata !== undefined || - tokens.has(event.eventData.token) || - ids.has(event.correlationId) - ) { - throw new WorkflowWorldError( - 'createHooks requires distinct, metadata-free system hooks', - { status: 400 } - ); - } - tokens.add(event.eventData.token); - ids.add(event.correlationId); - } - // One attempt. An ambiguous transport failure redelivers the suspension; - // it never falls back to individual writes while siblings may still commit. - const wire = await createWorkflowRunEventsBatchV4( - { - runId, - kind: 'hooks', - events: events.map(({ event, occurredAt }, index) => { - const { payload, meta } = splitEventDataForV4(event); - return { - runId, - eventType: event.eventType, - specVersion: event.specVersion ?? 2, - correlationId: event.correlationId, - occurredAt: occurredAt ?? new Date(), - ...(params?.requestId ? { vercelId: params.requestId } : {}), - ...(index === 0 && params?.replayDivergenceCount !== undefined - ? { replayDivergenceCount: params.replayDivergenceCount } - : {}), - ...(params?.eventCount !== undefined - ? { maxSlot: params.eventCount } - : {}), - remoteRefBehavior: 'resolve' as const, - payload, - ...meta, - }; - }), - }, - config - ); - return { - results: wire.results.map((item) => - item.error !== undefined - ? item - : { - status: 200 as const, - result: item, - } - ), - ...(wire.events !== undefined - ? { - events: wire.events, - cursor: wire.cursor, - hasMore: wire.hasMore, - } - : {}), - }; -} - export async function createWorkflowRunEvent( id: string | null, data: T, @@ -886,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); @@ -1015,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-vercel/src/instrumentObject.test.ts b/packages/world-vercel/src/instrumentObject.test.ts index 96594601f6..d7ac666989 100644 --- a/packages/world-vercel/src/instrumentObject.test.ts +++ b/packages/world-vercel/src/instrumentObject.test.ts @@ -84,29 +84,4 @@ describe('instrumentObject', () => { 'workflow.run.id': 'wrun_test', }); }); - it('tags system hook batches with their run and composition', async () => { - const events = instrumentObject('world.events', { - createHooks: vi.fn(async (_runId: string, _items: unknown[]) => ({ - results: [], - })), - }); - await events.createHooks('wrun_batch', [ - { event: { eventType: 'hook_created' } }, - { event: { eventType: 'hook_created' } }, - ]); - expect(mocks.trace).toHaveBeenCalledWith( - 'world.events.createHooks', - expect.anything(), - expect.any(Function) - ); - expect(mocks.span.setAttributes).toHaveBeenCalledWith( - expect.objectContaining({ 'workflow.run.id': 'wrun_batch' }) - ); - expect(mocks.span.setAttributes).toHaveBeenCalledWith( - expect.objectContaining({ - 'workflow.batch.size': 2, - 'workflow.batch.shape': 'hook_created:2', - }) - ); - }); }); diff --git a/packages/world-vercel/src/instrumentObject.ts b/packages/world-vercel/src/instrumentObject.ts index a1a239bbc4..0162d9071c 100644 --- a/packages/world-vercel/src/instrumentObject.ts +++ b/packages/world-vercel/src/instrumentObject.ts @@ -25,7 +25,6 @@ const RUN_ID_ARG_INDEX_BY_METHOD: Record = { 'world.runs.experimentalSetAttributes': 0, 'world.steps.get': 0, 'world.events.create': 0, - 'world.events.createHooks': 0, 'world.events.get': 0, }; @@ -142,10 +141,7 @@ export function instrumentObject(prefix: string, o: T): T { // event. These live here (the world.events.createBatch span), not on // the transport span underneath. let batchAttributes: Record | undefined; - if ( - prefix === 'world.events' && - (methodName === 'createBatch' || methodName === 'createHooks') - ) { + if (prefix === 'world.events' && methodName === 'createBatch') { const events = args[1]; if (Array.isArray(events)) { const counts = new Map(); diff --git a/packages/world-vercel/src/storage-hook-batch.test.ts b/packages/world-vercel/src/storage-hook-batch.test.ts deleted file mode 100644 index a32538c9ca..0000000000 --- a/packages/world-vercel/src/storage-hook-batch.test.ts +++ /dev/null @@ -1,96 +0,0 @@ -import { Buffer } from 'node:buffer'; -import { encode } from 'cbor-x'; -import { MockAgent } from 'undici'; -import { describe, expect, it } from 'vitest'; -import { encodeFrame, V4_FRAME_CONTENT_TYPE } from './frames.js'; -import { createStorage } from './storage.js'; -import { WORKFLOW_SERVER_URL_OVERRIDE } from './utils.js'; - -const ORIGIN = WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; - -function fixture() { - const agent = new MockAgent(); - agent.disableNetConnect(); - const storage = createStorage({ token: 'test', dispatcher: agent }); - const list = async (version?: string) => { - agent - .get(ORIGIN) - .intercept({ path: /\/api\/v4\/runs\/wrun_1\/events\?/, method: 'GET' }) - .reply( - 200, - Buffer.from(encodeFrame({ _end: 1, hasMore: false }, new Uint8Array())), - { - headers: { - 'content-type': V4_FRAME_CONTENT_TYPE, - ...(version === undefined - ? {} - : { 'x-wf-hook-batch-version': version }), - }, - } - ); - await storage.events.list({ runId: 'wrun_1' }); - }; - return { agent, storage, list }; -} - -describe('server-negotiated system hook batching', () => { - it('keeps individual registrations before setup and on older servers', async () => { - const { agent, storage, list } = fixture(); - try { - expect(storage.events.createHooks).toBeUndefined(); - await list(); - expect(storage.events.createHooks).toBeUndefined(); - agent.assertNoPendingInterceptors(); - } finally { - await agent.close(); - } - }); - - it('enables only the supported version and revokes it after a rollback', async () => { - const { agent, storage, list } = fixture(); - const other = createStorage(); - try { - await list('1'); - expect(storage.events.createHooks).toBeTypeOf('function'); - expect(other.events.createHooks).toBeUndefined(); - await list('2'); - expect(storage.events.createHooks).toBeUndefined(); - await list('1'); - expect(storage.events.createHooks).toBeTypeOf('function'); - await list(); - expect(storage.events.createHooks).toBeUndefined(); - agent.assertNoPendingInterceptors(); - } finally { - await agent.close(); - } - }); - - it('never falls back to individual writes after a batch is sent', async () => { - const { agent, storage, list } = fixture(); - try { - await list('1'); - agent - .get(ORIGIN) - .intercept({ - path: '/api/v4/runs/wrun_1/events/hooks/batch', - method: 'POST', - }) - .reply(404, encode({ error: 'not-found', message: 'Route missing' })); - await expect( - storage.events.createHooks?.( - 'wrun_1', - [1, 2].map((i) => ({ - event: { - eventType: 'hook_created' as const, - correlationId: `hook_${i}`, - eventData: { token: `abrt_${i}`, isSystem: true }, - }, - })) - ) - ).rejects.toThrow('Route missing'); - agent.assertNoPendingInterceptors(); - } finally { - await agent.close(); - } - }); -}); diff --git a/packages/world-vercel/src/storage.ts b/packages/world-vercel/src/storage.ts index 0ace9db5b8..0c81d4d0d5 100644 --- a/packages/world-vercel/src/storage.ts +++ b/packages/world-vercel/src/storage.ts @@ -6,7 +6,6 @@ import type { import { createWorkflowRunEvent, createWorkflowRunEventBatch, - createWorkflowRunHooks, getEvent, getWorkflowRunEvents, } from './events.js'; @@ -25,13 +24,6 @@ import { getStep, listWorkflowRunSteps } from './steps.js'; import type { APIConfig } from './utils.js'; export function createStorage(config?: APIConfig): Storage { - let hookBatchSupported = false; - config = { - ...config, - onHookBatchSupport: (supported) => { - hookBatchSupported = supported; - }, - }; const snapshots = createSnapshotsStorage(config); const storage: Storage = { // Storage interface with namespaced methods @@ -66,8 +58,6 @@ export function createStorage(config?: APIConfig): Storage { ) => createWorkflowRunEvent(runId, data, params, config), createBatch: (runId, events, params) => createWorkflowRunEventBatch(runId, events, params, config), - createHooks: (runId, events, params) => - createWorkflowRunHooks(runId, events, params, config), get: (runId, eventId, params) => getEvent(runId, eventId, params, config), list: (params) => getWorkflowRunEvents(params, config), listByCorrelationId: (params) => getWorkflowRunEvents(params, config), @@ -81,20 +71,11 @@ export function createStorage(config?: APIConfig): Storage { }; // Instrument all storage methods with tracing - const events = instrumentObject('world.events', storage.events); - const createHooks = events.createHooks; - // Keep the capability dynamic: replay responses can enable it, while an - // older/rolled-back server disables it on the next setup response. Do not - // retry individual writes after an ambiguous batch failure. - Object.defineProperty(events, 'createHooks', { - enumerable: true, - get: () => (hookBatchSupported ? createHooks : undefined), - }); // NOTE: Span names are lowercase per OTEL semantic conventions return { runs: instrumentObject('world.runs', storage.runs), steps: instrumentObject('world.steps', storage.steps), - events, + events: instrumentObject('world.events', storage.events), hooks: instrumentObject('world.hooks', storage.hooks), experimental_snapshots: instrumentObject( 'world.experimental_snapshots', diff --git a/packages/world-vercel/src/trace-propagation.test.ts b/packages/world-vercel/src/trace-propagation.test.ts index 9457750abc..6bd99a44e7 100644 --- a/packages/world-vercel/src/trace-propagation.test.ts +++ b/packages/world-vercel/src/trace-propagation.test.ts @@ -22,7 +22,6 @@ import { vi, } from 'vitest'; import { z } from 'zod'; -import { createWorkflowRunHooks } from './events.js'; import { createHookReceivedPreloadEventV4, getWorkflowRunEventsV4, @@ -611,52 +610,3 @@ describe.skipIf(WORKFLOW_SERVER_URL_OVERRIDE !== '')( }); } ); - -describe('system hook batch trace propagation', () => { - it('parents the shared registration HTTP request to the active workflow trace', async () => { - const fetchMock = vi.fn().mockResolvedValue( - cborResponse({ - results: [ - { status: 409, error: 'conflict', message: 'already registered' }, - { status: 409, error: 'conflict', message: 'already registered' }, - ], - }) - ); - vi.stubGlobal('fetch', fetchMock); - const tracer = otelTrace.getTracer('test'); - let traceId = ''; - let parentId = ''; - await tracer.startActiveSpan('flow-invocation', async (span) => { - traceId = span.spanContext().traceId; - parentId = span.spanContext().spanId; - await createWorkflowRunHooks( - 'wrun_trace', - ['cancel', 'steer'].map((name) => ({ - event: { - eventType: 'hook_created' as const, - specVersion: 6, - correlationId: `hook_${name}`, - eventData: { - token: `abrt_${name}`, - isSystem: true, - isWebhook: false, - }, - }, - })), - undefined, - { token: 'test-token' } - ); - span.end(); - }); - expect(fetchMock).toHaveBeenCalledTimes(1); - expect(String(fetchMock.mock.calls[0][0])).toContain('/events/hooks/batch'); - const headers = new Headers(fetchMock.mock.calls[0][1]?.headers); - const client = exporter - .getFinishedSpans() - .find((s) => s.name === 'http POST'); - expect(client?.parentSpanId).toBe(parentId); - expect(headers.get('traceparent')).toBe( - `00-${traceId}-${client?.spanContext().spanId}-01` - ); - }); -}); diff --git a/packages/world-vercel/src/utils.ts b/packages/world-vercel/src/utils.ts index ff81666683..3cba7d0808 100644 --- a/packages/world-vercel/src/utils.ts +++ b/packages/world-vercel/src/utils.ts @@ -140,12 +140,6 @@ export interface APIConfig { * @internal */ mintedSpecVersion?: number; - /** - * Observe server support on replay responses, before scheduling writes. - * Scoped to one storage instance; absent headers disable hook batching. - * @internal - */ - onHookBatchSupport?: (supported: boolean) => void; } export const DEFAULT_RESOLVE_DATA_OPTION = 'all'; diff --git a/packages/world/README.md b/packages/world/README.md index e46973df46..1156af9047 100644 --- a/packages/world/README.md +++ b/packages/world/README.md @@ -6,16 +6,6 @@ This package defines the `World` interface that abstracts workflow storage, queu Used internally by `@workflow/core` and world implementations. Should not be used directly in application code. -## Optional system hook batches - -`events.createHooks(runId, registrations, params?)` coalesces 2–32 distinct, -metadata-free generated system hook registrations. Its per-item outcomes -represent independent commits, not a group transaction. Implementations must -settle all launched writes before reporting failure. When returning a shared -history suffix, use the caller's `eventCount` as the slot frontier and return -complete ordered history or accurate pagination. An incomplete suffix must -never be presented as complete. Omit the method to retain individual writes. - ## Stream writer transport lifetime A stateful `StreamWriteSession` may implement `release()` to retire an idle @@ -63,6 +53,16 @@ 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. + ## 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 d37ce4d430..be23c364d7 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -1075,6 +1075,14 @@ 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. + */ + eventCursor?: string; /** The workflow run entity (for run_* events) */ run?: WorkflowRun; /** The step entity (for step_* events) */ @@ -1262,33 +1270,6 @@ export type BatchEventItemResult = wait?: undefined; }; -/** Parameters for coalescing independent system-hook registrations. */ -export type CreateHookBatchParams = Pick< - CreateEventParams, - 'eventCount' | 'requestId' | 'replayDivergenceCount' ->; - -/** - * Hook registrations commit independently, with the same per-hook atomicity - * and retry semantics as `create`. This is request coalescing, not a transaction - * across hooks (hook storage and the run log may live in different regions). - */ -export interface HookBatchResult { - results: ( - | { - status: 200; - error?: undefined; - message?: undefined; - result: EventResult; - } - | { status: number; error: string; message: string; result?: undefined } - )[]; - /** One shared, dense suffix after the requested eventCount, when available. */ - events?: Event[]; - cursor?: string | null; - hasMore?: boolean; -} - /** Result of {@link Storage.events.createBatch}. */ export interface EventBatchResult { /** One entry per submitted event, in request order. */ diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts index dd469629aa..cf8a419e78 100644 --- a/packages/world/src/interfaces.ts +++ b/packages/world/src/interfaces.ts @@ -8,13 +8,11 @@ import type { CreateEventBatchParams, CreateEventParams, CreateEventRequest, - CreateHookBatchParams, Event, EventBatchResult, EventListResponse, EventResult, GetEventParams, - HookBatchResult, ListEventsByCorrelationIdParams, ListEventsParams, RunCreatedEventRequest, @@ -468,22 +466,6 @@ export interface Storage { params?: CreateEventBatchParams ): Promise; - /** - * Optional request coalescing for distinct, generated system hooks. Each - * registration uses `create`'s lifecycle protocol; the group is NOT atomic. - * All writes settle before returning or rejecting. A shared history suffix - * avoids each registration reading back its siblings' events. Worlds - * without this method keep using individual registrations. - * - * Only hook_created, isSystem:true, non-webhook, non-force, metadata-free - * registrations with distinct tokens and correlation IDs are accepted. - */ - createHooks?( - runId: string, - events: BatchEventRequest[], - params?: CreateHookBatchParams - ): Promise; - get( runId: string, eventId: string, From 245e5946913c3ce44bfc514dbcfca69d09536c89 Mon Sep 17 00:00:00 2001 From: Karthik Kalyanaraman Date: Mon, 19 Oct 2026 05:52:12 -0700 Subject: [PATCH 4/4] Harden concurrent hook reconciliation and document its World contract Signed-off-by: Karthik Kalyanaraman --- .../docs/v5/how-it-works/event-sourcing.mdx | 2 +- docs/content/worlds/v5/building-a-world.mdx | 12 +- .../concurrent-hook-registration.test.ts | 121 +++++++++++++++--- packages/core/src/runtime/helpers.ts | 2 +- .../core/src/runtime/suspension-handler.ts | 117 ++++++++++------- packages/world-vercel/README.md | 19 ++- packages/world-vercel/src/events.test.ts | 107 ++++++++++++++++ packages/world/README.md | 6 + packages/world/src/events.ts | 10 +- 9 files changed, 328 insertions(+), 68 deletions(-) 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 0cf0257e85..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` for a lone hook; independent concurrent Node.js registrations 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_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/concurrent-hook-registration.test.ts b/packages/core/src/runtime/concurrent-hook-registration.test.ts index b692eec6e9..95bb6d4022 100644 --- a/packages/core/src/runtime/concurrent-hook-registration.test.ts +++ b/packages/core/src/runtime/concurrent-hook-registration.test.ts @@ -1,3 +1,4 @@ +import type { Span } from '@opentelemetry/api'; import { EntityConflictError, PreconditionFailedError, @@ -75,6 +76,7 @@ function setup(count = 2) { events: { create, list: vi.fn() }, getEncryptionKeyForRun: vi.fn(async () => undefined), } as unknown as World; + const span = { setAttributes: vi.fn() }; const invoke = ( options: Partial[0]> = {} ) => @@ -82,11 +84,12 @@ function setup(count = 2) { world, run, eventLog, + span: span as unknown as Span, requestId: 'request', suspension: new WorkflowSuspension(pending, globalThis), ...options, }); - return { create, answer, world, seed, eventLog, pending, invoke }; + return { create, answer, world, span, seed, eventLog, pending, invoke }; } describe('concurrent hook history reconciliation', () => { @@ -119,6 +122,11 @@ describe('concurrent hook history reconciliation', () => { ); 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', @@ -207,21 +215,21 @@ describe('concurrent hook history reconciliation', () => { }); it.each([ - 'gap before', - 'gap between', - 'missing event', - 'missing cursor', - 'duplicate slot', - 'old event', - 'wrong run', - 'wrong correlation', - 'wrong token', - 'wrong type', - 'non-slot ID', - 'truncated page', - 'lazy metadata', - 'omitted metadata', - ])('retains the reload path without changing the log: %s', async (mode) => { + ['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' }); @@ -259,6 +267,11 @@ describe('concurrent hook history reconciliation', () => { 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 () => { @@ -267,6 +280,11 @@ describe('concurrent hook history reconciliation', () => { 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 () => { @@ -288,6 +306,11 @@ describe('concurrent hook history reconciliation', () => { }); 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 () => { @@ -350,9 +373,75 @@ describe('concurrent hook history reconciliation', () => { (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', diff --git a/packages/core/src/runtime/helpers.ts b/packages/core/src/runtime/helpers.ts index 349f1f7c1c..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); independent concurrent node registrations defer both | a lone create returns a delta; concurrent registrations reconcile committed responses together, reloading when they do not prove a consecutive suffix | + * | `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 75033aeb03..24893dfe58 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -808,16 +808,28 @@ export async function handleSuspension({ // 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. - const concurrentHooks = - eventLog && (run.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_SLOT_IDENTITY - ? hooksNeedingCreation.filter( - (item) => - !item.force && - !item.disposed && - !item.abortRequested && - hookItemsByToken.get(item.token)?.length === 1 - ) - : []; + // 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) @@ -1997,53 +2009,72 @@ export async function handleSuspension({ // 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. - let concurrentHooksCarriedForward = false; - if ( - eventLog && - hookFrontier !== undefined && - concurrentHookIds.size > 1 && - hookResults.size === concurrentHookIds.size && - eventWrites === hookResults.size && - maxEventSlot(eventLog.events) === hookFrontier && - findEventSlotGap(eventLog.events) === undefined - ) { - const committed: { event: Event; cursor?: string }[] = []; + 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 || - result.hasMore === true || event.runId !== runId || event.correlationId !== item.correlationId || (event.eventType !== 'hook_created' && event.eventType !== 'hook_conflict') || - event.eventData.token !== item.token || - (event.eventType === 'hook_created' && - ((event.eventData.metadata !== undefined && - !(event.eventData.metadata instanceof Uint8Array)) || - (item.metadata !== undefined && - event.eventData.metadata === undefined))) || - eventIdToSlot(event.eventId) === null + 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)) ) { - break; + return 'unresolved_metadata'; } - committed.push({ event, cursor: result.eventCursor }); + const slot = eventIdToSlot(event.eventId); + if (slot === null) return 'non_slot_event'; + committed.push({ event, slot, cursor: result.eventCursor }); } - committed.sort((a, b) => a.event.eventId.localeCompare(b.event.eventId)); - const last = committed.at(-1); + committed.sort((a, b) => a.slot - b.slot); if ( - committed.length === concurrentHookIds.size && - last?.cursor && - committed.every( - ({ event }, index) => - eventIdToSlot(event.eventId) === hookFrontier + index + 1 - ) + !committed.every(({ slot }, index) => slot === hookFrontier + index + 1) ) { - eventLog.events.push(...committed.map(({ event }) => event)); - eventLog.cursor = last.cursor; - concurrentHooksCarriedForward = true; + 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 diff --git a/packages/world-vercel/README.md b/packages/world-vercel/README.md index 2f713f0dd2..2eea915c2c 100644 --- a/packages/world-vercel/README.md +++ b/packages/world-vercel/README.md @@ -1,5 +1,13 @@ # @workflow/world-vercel +Production workflow backend for Vercel platform deployments. + +Integrates with Vercel's infrastructure for storage, queuing, and authentication. Handles workflow persistence and scaling in production environments. + +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 @@ -8,11 +16,12 @@ 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. -Production workflow backend for Vercel platform deployments. - -Integrates with Vercel's infrastructure for storage, queuing, and authentication. Handles workflow persistence and scaling in production environments. - -Used by default for deployments on Vercel. Authentication and API endpoints are configured automatically in Vercel deployments. +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 diff --git a/packages/world-vercel/src/events.test.ts b/packages/world-vercel/src/events.test.ts index fa39146e50..1b960146d3 100644 --- a/packages/world-vercel/src/events.test.ts +++ b/packages/world-vercel/src/events.test.ts @@ -37,6 +37,113 @@ 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[] = []; diff --git a/packages/world/README.md b/packages/world/README.md index 1156af9047..7bf345ce53 100644 --- a/packages/world/README.md +++ b/packages/world/README.md @@ -63,6 +63,12 @@ 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 be23c364d7..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 @@ -1081,6 +1085,10 @@ export type EventResult = { * 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) */