From 41f460f8d265143b6853983f907564f339497578 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Thu, 13 Aug 2026 21:20:16 -0700 Subject: [PATCH 1/6] =?UTF-8?q?feat(world,world-vercel):=20createBatch=20?= =?UTF-8?q?=E2=80=94=20ordered=20batch=20event=20write=20with=20per-event?= =?UTF-8?q?=20results?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The client half of workflow-server#646's v4 batch endpoint, rebuilt from scratch against the merged contract (this branch's previous fence-based design is retired; old head tagged batch-client-v2-fence-design). - @workflow/world: optional `events.createBatch(runId, events, params)` — ordered events in, per-event outcomes out. Method presence IS the capability declaration: worlds that don't implement it keep the single-event path (world-local and world-postgres stamp specVersion 5 and gain nothing from batching a local write, so they deliberately don't implement it). Each result reports what that event's own single create would have returned: 200 + the materialized entity, or the single-path status/code (409 conflict for an already-applied event), so callers reuse single-path conflict handling per event. - @workflow/world-vercel: POST /v4/runs/:runId/events/batch — the events' single-POST frames back-to-back (byte-identical framing, no batch-level meta, no fence fields), CBOR { results } decoded against the SAME per-type schemas as the single POST, with a loud SCHEMA_VALIDATION on a malformed response (length mismatch, invalid item). Wired into createStorage. - Retry: the whole batch POST is idempotent-on-retry regardless of the event types it carries (every batchable event is guarded by its own entity condition; a retried committed batch converges to per-event 409s), so a new batchIdempotent override joins the per-type eligibility matrix — including #3504's in-process 429 Retry-After handling. Runtime integration (suspension fan-out fold, then the deferred sequential transition) ships separately on top of this contract — see the PR description for the staged plan, caps, and kill switch. Tests: 7 wire tests (frame encoding + ordering + no fence fields, per-event result mapping, malformed-response failures, typed 400s, 5xx in-process retry, empty-batch guard); world-vercel suite 508/508. Co-Authored-By: Claude Fable 5 --- packages/world-vercel/src/event-retry.ts | 13 + .../world-vercel/src/events-batch.test.ts | 385 ++++++++++++++++++ packages/world-vercel/src/events-v4.ts | 126 ++++++ packages/world-vercel/src/events.ts | 71 ++++ packages/world-vercel/src/storage.ts | 3 + packages/world/src/events.ts | 63 +++ packages/world/src/interfaces.ts | 30 ++ 7 files changed, 691 insertions(+) create mode 100644 packages/world-vercel/src/events-batch.test.ts diff --git a/packages/world-vercel/src/event-retry.ts b/packages/world-vercel/src/event-retry.ts index a8cd8f72a0..af8daa7b5a 100644 --- a/packages/world-vercel/src/event-retry.ts +++ b/packages/world-vercel/src/event-retry.ts @@ -331,6 +331,18 @@ export interface EventPostRetryOptions { * responses stay non-retryable regardless. */ idempotentHookResume?: boolean; + /** + * Batch POST override: a `createBatch` request is idempotent-on-retry as a + * WHOLE regardless of the event types it carries, because every batchable + * event is guarded by its own entity condition — a retry of a batch that + * committed (or partially committed) converges to per-event 409 results + * with nothing written twice, which the caller already handles per event. + * The per-type eligibility matrix guards SINGLE posts, where e.g. a + * retried `step_started` would increment `attempt` unconditionally; in a + * batch that same start is fenced by the step create-claim, so the matrix + * does not apply. Definitive 4xx responses stay non-retryable regardless. + */ + batchIdempotent?: boolean; } /** @@ -376,6 +388,7 @@ function isEligibleForTransientRetry( options?: EventPostRetryOptions ): boolean { return ( + options?.batchIdempotent === true || (eventType === 'hook_received' && options?.idempotentHookResume === true) || (EVENT_RETRY_ELIGIBILITY[eventType]?.retryable ?? false) ); diff --git a/packages/world-vercel/src/events-batch.test.ts b/packages/world-vercel/src/events-batch.test.ts new file mode 100644 index 0000000000..9b75af0179 --- /dev/null +++ b/packages/world-vercel/src/events-batch.test.ts @@ -0,0 +1,385 @@ +import { Buffer } from 'node:buffer'; +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 { WORKFLOW_SERVER_URL_OVERRIDE } from './utils.js'; + +/** + * POST /api/v4/runs/:runId/events/batch — the client wire half. + * + * What these tests pin down: + * - the request body is the events' single-POST frames back-to-back, in + * request order, with per-event meta and payload bytes and NO batch-level + * fence fields (the retired v2 design's expectedRunVersion / batchId / + * logicalCreatedAt must never reappear on the wire); + * - per-event results map through typed: successes validate against the + * same per-type schema as the single POST, failures pass through + * { status, error, message } untouched; + * - a malformed response (wrong results length, invalid item body) fails + * loudly as SCHEMA_VALIDATION rather than degrading into per-event + * failures; + * - the whole POST is idempotent-on-retry (batchIdempotent): a transient + * 5xx is retried in-process regardless of the contained event types. + */ + +const ORIGIN = WORKFLOW_SERVER_URL_OVERRIDE || 'https://vercel-workflow.com'; +const CREATED_AT = '2026-08-14T00:00:00.000Z'; +const RUN_ID = 'wrun_1'; + +const utf8 = (value: string): Uint8Array => new TextEncoder().encode(value); + +/** Slot-identity event id for `slot` (26-char zero-padded decimal body). */ +const slotEventId = (slot: number): string => + `evnt_${String(slot).padStart(26, '0')}`; + +function mockAgent(): MockAgent { + const agent = new MockAgent(); + agent.disableNetConnect(); + return agent; +} + +/** Parse the concatenated v4 frames of a batch request body. */ +function decodeBatchFrames( + body: Uint8Array +): { meta: Record; payload: Uint8Array }[] { + const frames: { meta: Record; payload: Uint8Array }[] = []; + const view = new DataView(body.buffer, body.byteOffset, body.byteLength); + let offset = 0; + while (offset < body.byteLength) { + const metaLen = view.getUint32(offset, false); + const meta = decode(body.subarray(offset + 4, offset + 4 + metaLen)); + const bodyLen = view.getUint32(offset + 4 + metaLen, false); + const payload = body.subarray( + offset + 4 + metaLen + 4, + offset + 4 + metaLen + 4 + bodyLen + ); + frames.push({ meta, payload }); + offset += 4 + metaLen + 4 + bodyLen; + } + return frames; +} + +/** The sequential-transition batch: completed(A), created(B), started(B). */ +function transitionEvents(): BatchEventRequest[] { + return [ + { + event: { + eventType: 'step_completed', + specVersion: 6, + correlationId: 'step_a', + eventData: { + stepName: 'step-a', + workflowName: 'wf', + result: utf8('"a-output"'), + }, + }, + occurredAt: new Date('2026-08-14T00:00:01.000Z'), + }, + { + event: { + eventType: 'step_created', + specVersion: 6, + correlationId: 'step_b', + eventData: { + stepName: 'step-b', + workflowName: 'wf', + input: utf8('"b-input"'), + }, + }, + occurredAt: new Date('2026-08-14T00:00:02.000Z'), + }, + { + event: { + eventType: 'step_started', + specVersion: 6, + correlationId: 'step_b', + eventData: { stepName: 'step-b' }, + }, + occurredAt: new Date('2026-08-14T00:00:03.000Z'), + }, + ]; +} + +const completedEvent = { + eventId: slotEventId(8), + runId: RUN_ID, + eventType: 'step_completed', + correlationId: 'step_a', + createdAt: CREATED_AT, + eventData: { stepName: 'step-a' }, +}; + +const stepA = { + runId: RUN_ID, + stepId: 'step_a', + stepName: 'step-a', + status: 'completed', + attempt: 1, + createdAt: CREATED_AT, + updatedAt: CREATED_AT, +}; + +const createdEvent = { + eventId: slotEventId(9), + runId: RUN_ID, + eventType: 'step_created', + correlationId: 'step_b', + createdAt: CREATED_AT, + eventData: { stepName: 'step-b' }, +}; + +const startedEvent = { + eventId: slotEventId(10), + runId: RUN_ID, + eventType: 'step_started', + correlationId: 'step_b', + createdAt: CREATED_AT, + eventData: { stepName: 'step-b' }, +}; + +const stepB = { + runId: RUN_ID, + stepId: 'step_b', + stepName: 'step-b', + status: 'running', + attempt: 1, + createdAt: CREATED_AT, + updatedAt: CREATED_AT, + startedAt: CREATED_AT, +}; + +const fullSuccessBody = () => + encode({ + results: [ + { status: 200, event: completedEvent, step: stepA }, + { status: 200, event: createdEvent, step: { ...stepB } }, + { status: 200, event: startedEvent, step: { ...stepB } }, + ], + }); + +describe('createWorkflowRunEventBatch', () => { + it('encodes one single-POST frame per event, in order, with no fence fields', async () => { + const agent = mockAgent(); + let requestBody: Uint8Array | undefined; + agent + .get(ORIGIN) + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/batch`, + method: 'POST', + body: (raw) => { + requestBody = new Uint8Array(Buffer.from(raw, 'binary')); + return true; + }, + }) + .reply(200, fullSuccessBody(), { + headers: { 'content-type': 'application/cbor' }, + }); + + const result = await createWorkflowRunEventBatch( + RUN_ID, + transitionEvents(), + undefined, + { token: 'test-token', dispatcher: agent } + ); + + expect(result.results).toHaveLength(3); + expect(requestBody).toBeDefined(); + // biome-ignore lint/style/noNonNullAssertion: asserted above + const frames = decodeBatchFrames(requestBody!); + expect(frames).toHaveLength(3); + expect(frames.map((frame) => frame.meta.eventType)).toEqual([ + 'step_completed', + 'step_created', + 'step_started', + ]); + expect(frames.map((frame) => frame.meta.correlationId)).toEqual([ + 'step_a', + 'step_b', + 'step_b', + ]); + // Payload bytes ride the frame body; step_started has none. + expect(new TextDecoder().decode(frames[0].payload)).toBe('"a-output"'); + expect(new TextDecoder().decode(frames[1].payload)).toBe('"b-input"'); + expect(frames[2].payload.byteLength).toBe(0); + // Each frame carries its own client event time (the source of the + // durable createdAt under slot identity). + for (const frame of frames) { + expect(frame.meta.occurredAt).toBeInstanceOf(Date); + } + // The retired fence design must never reappear on the wire. + for (const frame of frames) { + expect(frame.meta).not.toHaveProperty('expectedRunVersion'); + expect(frame.meta).not.toHaveProperty('batchId'); + expect(frame.meta).not.toHaveProperty('logicalCreatedAt'); + } + agent.assertNoPendingInterceptors(); + }); + + it('maps per-event outcomes: successes typed, failures passed through', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/batch`, + method: 'POST', + }) + .reply( + 200, + encode({ + results: [ + { + status: 409, + error: 'conflict', + message: 'step step_a already completed', + }, + { status: 200, event: createdEvent, step: { ...stepB } }, + { status: 200, event: startedEvent, step: { ...stepB } }, + ], + }), + { headers: { 'content-type': 'application/cbor' } } + ); + + const { results } = await createWorkflowRunEventBatch( + RUN_ID, + transitionEvents(), + undefined, + { token: 'test-token', dispatcher: agent } + ); + + expect(results[0]).toEqual({ + status: 409, + error: 'conflict', + message: 'step step_a already completed', + }); + expect(results[1].status).toBe(200); + expect(results[1].event?.eventId).toBe(slotEventId(9)); + expect(results[2].status).toBe(200); + expect(results[2].step?.status).toBe('running'); + agent.assertNoPendingInterceptors(); + }); + + it('fails loudly when results length does not match the submitted frames', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/batch`, + method: 'POST', + }) + .reply( + 200, + encode({ + results: [{ status: 200, event: completedEvent, step: stepA }], + }), + { headers: { 'content-type': 'application/cbor' } } + ); + + await expect( + createWorkflowRunEventBatch(RUN_ID, transitionEvents(), undefined, { + token: 'test-token', + dispatcher: agent, + }) + ).rejects.toMatchObject({ + name: 'WorkflowWorldError', + code: 'SCHEMA_VALIDATION', + }); + }); + + it('fails loudly on an invalid success-item body, naming the index', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/batch`, + method: 'POST', + }) + .reply( + 200, + encode({ + results: [ + { status: 200, event: completedEvent, step: stepA }, + { status: 200 }, // missing event — protocol violation + { status: 200, event: startedEvent, step: { ...stepB } }, + ], + }), + { headers: { 'content-type': 'application/cbor' } } + ); + + await expect( + createWorkflowRunEventBatch(RUN_ID, transitionEvents(), undefined, { + token: 'test-token', + dispatcher: agent, + }) + ).rejects.toMatchObject({ + code: 'SCHEMA_VALIDATION', + message: expect.stringContaining('index 1'), + }); + }); + + it('maps a request-level 400 to a typed WorkflowWorldError', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/batch`, + method: 'POST', + }) + .reply( + 400, + JSON.stringify({ + success: false, + error: 'invalid-event-batch', + message: 'boom', + }), + { headers: { 'content-type': 'application/json' } } + ); + + await expect( + createWorkflowRunEventBatch(RUN_ID, transitionEvents(), undefined, { + token: 'test-token', + dispatcher: agent, + }) + ).rejects.toMatchObject({ name: 'WorkflowWorldError', status: 400 }); + }); + + it('retries a transient 5xx in-process (the batch POST is idempotent-on-retry)', async () => { + const agent = mockAgent(); + const pool = agent.get(ORIGIN); + pool + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/batch`, + method: 'POST', + }) + .reply(503, JSON.stringify({ message: 'unavailable' }), { + headers: { 'content-type': 'application/json' }, + }); + pool + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/batch`, + method: 'POST', + }) + .reply(200, fullSuccessBody(), { + headers: { 'content-type': 'application/cbor' }, + }); + + const { results } = await createWorkflowRunEventBatch( + RUN_ID, + transitionEvents(), + undefined, + { token: 'test-token', dispatcher: agent } + ); + expect(results.map((result) => result.status)).toEqual([200, 200, 200]); + agent.assertNoPendingInterceptors(); + }); + + it('rejects an empty batch without touching the network', async () => { + await expect( + createWorkflowRunEventBatch(RUN_ID, [], undefined, { + token: 'test-token', + dispatcher: mockAgent(), + }) + ).rejects.toMatchObject({ status: 400 }); + }); +}); diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index 93dd6ba96f..52d8a6d603 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -793,6 +793,132 @@ export async function createWorkflowRunStartedEventV4( return { events, ...page, maxEvents: maxEvents.data }; } +/** One event of a v4 batch POST, index-aligned with the response results. */ +export type CreateEventBatchV4Event = CreateEventV4InputBase & { + eventType: EventType; +}; + +export interface CreateEventBatchV4Input { + runId: string; + /** Events in request order — the order they land in the run's log. */ + events: CreateEventBatchV4Event[]; +} + +/** + * One event's outcome in a batch response. `error === undefined` + * discriminates success; a success item carries the same materialized body + * its single-event POST would have returned (validated against the same + * per-type schema). + */ +export type CreateEventBatchV4ItemResult = + | ({ status: 200; error?: undefined; message?: undefined } & EventResult & + Record<'event', Event>) + | { status: number; error: string; message: string; event?: undefined }; + +export interface CreateEventBatchV4Result { + results: CreateEventBatchV4ItemResult[]; +} + +const BatchItemFailureSchema = z.object({ + status: z.number().int(), + error: z.string(), + message: z.string(), +}); + +/** + * 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 + * 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. Slot-identity runs only — an older server 404s the + * route and a pre-slot run is rejected with a 400, both of which callers + * treat as "fall back to single-event posts". + */ +export async function createWorkflowRunEventsBatchV4( + input: CreateEventBatchV4Input, + config?: APIConfig +): Promise { + assert(input.events.length > 0, 'v4 createEventBatch: empty batch'); + const { baseUrl, headers: baseHeaders } = await getHttpConfig(config); + const headers = new Headers(baseHeaders); + // Match the single-event POST content type — the batch route runs on the + // same authed + v4 middleware chain and the frame bytes are identical. + headers.set('Content-Type', 'application/octet-stream'); + + const frames = input.events.map((event) => + encodeFrame(buildPostFrameMeta(event), event.payload ?? new Uint8Array(0)) + ); + let total = 0; + for (const frame of frames) total += frame.byteLength; + const body = new Uint8Array(total); + let offset = 0; + for (const frame of frames) { + body.set(frame, offset); + offset += frame.byteLength; + } + + const url = `${baseUrl}/v4/runs/${encodeURIComponent(input.runId)}/events/batch`; + const response = await fetchV4( + url, + { method: 'POST', headers, body }, + config, + 'createEventBatch', + { + ...WorkflowEventsTransport('http'), + ...WorkflowEventType(input.events[0].eventType), + 'workflow.batch.size': input.events.length, + } + ); + + const bodyBytes = new Uint8Array(await response.arrayBuffer()); + const decoded = + bodyBytes.byteLength > 0 + ? (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 / + // short `results` is a server protocol violation; silently coercing it + // would masquerade as per-event failures and hide the server bug. The + // batch POST is idempotent-on-retry (per-event entity conditions), so + // failing loudly here is safe for the retry wrapper to re-send. + if ( + !Array.isArray(decoded.results) || + decoded.results.length !== input.events.length + ) { + throw new WorkflowWorldError( + `v4 createEventBatch: response \`results\` length ` + + `(${Array.isArray(decoded.results) ? decoded.results.length : 'non-array'}) ` + + `!= ${input.events.length} submitted frames`, + { code: 'SCHEMA_VALIDATION' } + ); + } + + const results = decoded.results.map( + (raw, index): CreateEventBatchV4ItemResult => { + const failure = BatchItemFailureSchema.safeParse(raw); + if (failure.success && failure.data.status !== 200) { + return failure.data; + } + // Success items validate against the SAME per-type schema the single + // POST uses, so a batched write and its single-path twin return + // byte-equivalent bodies to the caller. + const eventType = input.events[index].eventType; + const parsed = CreateEventV4BodySchemas[eventType].safeParse(raw); + if (!parsed.success) { + throw new WorkflowWorldError( + `v4 createEventBatch: invalid result body at index ${index} (${eventType})`, + { code: 'SCHEMA_VALIDATION', cause: parsed.error } + ); + } + return { status: 200, ...parsed.data }; + } + ); + + return { results }; +} + /** The only two members a decoded transport result is read for. `fetch`'s * `Response` satisfies it structurally, so the HTTP branch returns one * unchanged and the WS branch synthesizes the same shape. */ diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index 8ecc213af6..79e959b5b8 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -35,8 +35,12 @@ import { HookNotFoundError, WorkflowWorldError } from '@workflow/errors'; import { type AnyEventRequest, applyAttributeChanges, + type BatchEventItemResult, + type BatchEventRequest, + type CreateEventBatchParams, type CreateEventParams, type Event, + type EventBatchResult, type EventDataPayloadField, type EventResult, EventSchema, @@ -52,6 +56,7 @@ import { import { withEventPostRetry } from './event-retry.js'; import { createHookReceivedPreloadEventV4, + createWorkflowRunEventsBatchV4, createWorkflowRunEventV4, createWorkflowRunStartedEventV4, getEventsByCorrelationIdV4, @@ -470,6 +475,72 @@ export async function getWorkflowRunEvents( }; } +/** + * Batch write: append an ordered list of events to the run's log in one + * request with per-event outcomes — the world-vercel implementation of + * `Storage['events']['createBatch']`. + * + * The whole POST retries transient transport failures and 429s like a single + * event write does, and is safe to: every batchable event is guarded by its + * own entity condition server-side, so a retry of a batch that (partially) + * committed converges to per-event 409 results with nothing written twice. + */ +export async function createWorkflowRunEventBatch( + runId: string, + events: BatchEventRequest[], + _params?: CreateEventBatchParams, + config?: APIConfig +): Promise { + if (events.length === 0) { + throw new WorkflowWorldError( + 'world-vercel: createBatch requires at least one event', + { status: 400 } + ); + } + const inputs = events.map(({ event, occurredAt }) => { + const { payload, meta } = splitEventDataForV4(event); + return { + runId, + eventType: event.eventType, + specVersion: event.specVersion ?? 2, + ...(event.correlationId ? { correlationId: event.correlationId } : {}), + // Under slot identity this is the source of the durable createdAt, so + // the caller's logical time is what every replay observes. + occurredAt: occurredAt ?? new Date(), + // Batch responses carry entities for bookkeeping, not payload reads — + // the caller just produced every payload in this batch itself. + remoteRefBehavior: 'lazy' as const, + payload, + ...meta, + }; + }); + + const wire = await withEventPostRetry( + () => createWorkflowRunEventsBatchV4({ runId, events: inputs }, config), + events[0].event.eventType, + { batchIdempotent: true } + ); + + return { + results: wire.results.map((item): BatchEventItemResult => { + if (item.error !== undefined) { + return { + status: item.status, + error: item.error, + message: item.message, + }; + } + return { + status: 200, + event: item.event, + ...(item.run ? { run: item.run } : {}), + ...(item.step ? { step: item.step } : {}), + ...(item.wait ? { wait: item.wait } : {}), + }; + }), + }; +} + export async function createWorkflowRunEvent( id: string | null, data: T, diff --git a/packages/world-vercel/src/storage.ts b/packages/world-vercel/src/storage.ts index 46fced038c..186c759805 100644 --- a/packages/world-vercel/src/storage.ts +++ b/packages/world-vercel/src/storage.ts @@ -5,6 +5,7 @@ import type { } from '@workflow/world'; import { createWorkflowRunEvent, + createWorkflowRunEventBatch, getEvent, getWorkflowRunEvents, } from './events.js'; @@ -48,6 +49,8 @@ export function createStorage(config?: APIConfig): Storage { data: AnyEventRequest, params?: CreateEventParams ) => createWorkflowRunEvent(runId, data, params, config), + createBatch: (runId, events, params) => + createWorkflowRunEventBatch(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/src/events.ts b/packages/world/src/events.ts index 63bd5afd74..08446b84f0 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -1017,6 +1017,69 @@ export type EventResult = { ? { step: StartedStep } : unknown); +/** + * One event of a batch write ({@link Storage.events.createBatch}), in request + * order — which is the order the events land in the run's log. + */ +export interface BatchEventRequest { + /** The event, same discriminated union the single `create` takes. */ + event: CreateEventRequest; + /** + * Client event time for this event. Under slot identity this is the source + * of the durable event's `createdAt` (a slot id carries no time), so the + * timestamp a replay observes is the one the writer chose — set it to the + * instant the event logically occurred. + */ + occurredAt?: Date; +} + +/** Per-batch parameters for {@link Storage.events.createBatch}. */ +export interface CreateEventBatchParams { + resolveData?: ResolveData; +} + +/** + * One event's outcome in a batch response, index-aligned with the submitted + * events. `error === undefined` discriminates success. + * + * A batch is processed as a whole (HTTP 200 whenever the World evaluated it); + * each event reports the outcome its OWN single `create` would have had: + * + * - success → `status: 200` plus the committed event and the same + * materialized entity the single create returns (`step` for step events, + * `wait` for wait events, `run` for run terminals); + * - rejection → the status code and error code the single create would have + * failed with (e.g. `409`/`conflict` for an event an earlier delivery + * already applied), so callers reuse their single-path conflict handling + * per event. A transport retry of a fully committed batch converges to + * all-409s with nothing written twice. + */ +export type BatchEventItemResult = + | { + status: 200; + error?: undefined; + message?: undefined; + event: Event; + run?: WorkflowRun; + step?: Step; + wait?: Wait; + } + | { + status: number; + error: string; + message: string; + event?: undefined; + run?: undefined; + step?: undefined; + wait?: undefined; + }; + +/** Result of {@link Storage.events.createBatch}. */ +export interface EventBatchResult { + /** One entry per submitted event, in request order. */ + results: BatchEventItemResult[]; +} + export interface GetEventParams { resolveData?: ResolveData; } diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts index ed76fd0c00..dd30a69b9c 100644 --- a/packages/world/src/interfaces.ts +++ b/packages/world/src/interfaces.ts @@ -4,9 +4,12 @@ import type { ExperimentalSetAttributesResult, } from './attributes.js'; import type { + BatchEventRequest, + CreateEventBatchParams, CreateEventParams, CreateEventRequest, Event, + EventBatchResult, EventResult, GetEventParams, ListEventsByCorrelationIdParams, @@ -323,6 +326,33 @@ export interface Storage { params?: CreateEventParams ): Promise>; + /** + * OPTIONAL batch write — append an ordered list of events to the run's + * log in one durable, atomic-per-attempt write, with a per-event outcome + * for each (see {@link BatchEventItemResult}). The events land in request + * order at consecutive slots (subject to bump-and-report semantics: a + * concurrent writer may push the whole batch to higher slots). + * + * Presence of the method IS the capability declaration: the core runtime + * batches only when the World implements it (and the run's spec version + * supports slot identity); absent, every write takes the single-event + * `create` path unchanged. A World must implement it with real + * atomicity per attempt — a lost race must leave nothing behind — or not + * implement it at all. + * + * Not expressible in a batch (Worlds reject the whole batch with a + * request-level error): `run_created`, `run_started`, `run_cancelled`, + * `hook_created`, `hook_disposed`, `attr_set`, and more events targeting + * one entity than a single write can express (the one legal combination + * is `step_created` followed by `step_started` for the same step, which + * creates the step born-running). + */ + createBatch?( + runId: string, + events: BatchEventRequest[], + params?: CreateEventBatchParams + ): Promise; + get( runId: string, eventId: string, From 9b9a5e2f4b0e6753d89758fbe4672294ae003c70 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 14 Aug 2026 13:54:55 -0700 Subject: [PATCH 2/6] feat(core): batched suspension fan-out via events.createBatch (opt-in) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit With WORKFLOW_BATCH_TRANSITIONS=1, the suspension handler folds a clean fan-out's eager step_created + wait_created writes into world.events.createBatch calls (one durable write, per-event outcomes) of at most MAX_BATCH_FANOUT_EVENTS (32) events, instead of one write per event. The fold engages only for a CLEAN fan-out: the World implements the optional createBatch, the run is on slot identity (specVersion >= 6), and the suspension carries no attribute writes, no hook writes, and no resilient step dispatch (whose creates are each paired with a queue publish). Everything outside the gate keeps the single-event path byte-for-byte. Lazy-inline steps keep deferring their step_created to the lazy start exactly as before. Per-event semantics mirror the single path: a 409 is the same already-exists tolerance as EntityConflictError (the conflicted step is not owned); any other per-event failure fails the suspension write the way a single-path rejection would. Slot bumps (the batch endpoint has no bump-and-report) are tolerated and logged — the same accepted exposure as a dropped truncated report on the single path. Off by default, staged like resilient step dispatch: opt-in for burn-in, then default-on with the variable retained as the kill switch. Docs: v5 changelog entry documenting the World spec addition (createBatch signature + contract) and a WORKFLOW_BATCH_TRANSITIONS entry in the worlds configuration reference. Tests: 9 new suspension-handler tests — the fold's ordering (steps then waits), per-event 409 tolerance, non-409 failure propagation, every gate exclusion (flag off, no createBatch, pre-slot run, hook writes), 32-cap chunking, and lazy-inline exclusion. Core suite: 2126 passed. Co-Authored-By: Claude Fable 5 --- .../v5/changelog/batched-event-writes.mdx | 58 +++ docs/content/docs/v5/changelog/meta.json | 3 +- docs/content/docs/v5/configuration/worlds.mdx | 8 + packages/core/src/runtime/constants.ts | 31 ++ .../src/runtime/suspension-handler.test.ts | 274 +++++++++++ .../core/src/runtime/suspension-handler.ts | 444 ++++++++++++------ 6 files changed, 673 insertions(+), 145 deletions(-) create mode 100644 docs/content/docs/v5/changelog/batched-event-writes.mdx diff --git a/docs/content/docs/v5/changelog/batched-event-writes.mdx b/docs/content/docs/v5/changelog/batched-event-writes.mdx new file mode 100644 index 0000000000..6f526156c0 --- /dev/null +++ b/docs/content/docs/v5/changelog/batched-event-writes.mdx @@ -0,0 +1,58 @@ +--- +title: Batched event writes +description: An optional World API (events.createBatch) that appends an ordered set of events in one durable write with per-event outcomes, and a suspension fan-out fold that uses it. +--- + +# Batched event writes (`events.createBatch`) + +## Motivation + +A workflow suspension that schedules several steps and waits previously wrote one event per entity — one `world.events.create` call per `step_created` and `wait_created`. Against a remote World each write is its own network round trip and its own crash boundary. Batching folds a suspension's schedule into **one durable write** with per-event outcomes, cutting request count and making the whole fan-out land atomically per attempt. + +## The World spec addition + +`Storage['events']` gains one **optional** method: + +```ts +createBatch?( + runId: string, + events: BatchEventRequest[], + params?: CreateEventBatchParams +): Promise; + +interface BatchEventRequest { + /** The event — the same discriminated union the single `create` takes. */ + event: CreateEventRequest; + /** Client event time; under slot identity, the source of the durable createdAt. */ + occurredAt?: Date; +} + +type BatchEventItemResult = + | { status: 200; event: Event; run?: WorkflowRun; step?: Step; wait?: Wait } + | { status: number; error: string; message: string }; + +interface EventBatchResult { + /** One entry per submitted event, in request order. */ + results: BatchEventItemResult[]; +} +``` + +The contract: + +- **Ordered**: events land in the run's log in request order (at consecutive slots, subject to bump-and-report semantics — a concurrent writer may push the whole batch to higher slots). +- **Per-event outcomes**: the batch is processed as a whole, and each event reports what its own single `create` would have returned — `200` plus the materialized entity, or the single-path status/code (`409`/`conflict` for an event an earlier delivery already applied). Callers reuse their single-path conflict handling per event. +- **Idempotent on retry**: every batchable event is guarded by its own entity condition, so retrying a batch that (partially) committed converges to per-event `409`s with nothing written twice. +- **Method presence is the capability declaration.** A World that doesn't implement it keeps the single-event path; a World that implements it must make each attempt atomic (a lost race leaves nothing behind). `world-vercel` implements it against `POST /v4/runs/:runId/events/batch` (slot-identity runs only, i.e. specVersion ≥ 6). `world-local` and `world-postgres` deliberately do not — batching buys nothing for a local write. +- **Not batchable** (Worlds reject the request): `run_created`, `run_started`, `run_cancelled`, `hook_created`, `hook_disposed`, `attr_set`, and multiple events targeting one entity — except `step_created` followed by `step_started` for the same step, which creates the step born-running. + +## The runtime integration (suspension fan-out fold) + +With `WORKFLOW_BATCH_TRANSITIONS=1`, the suspension handler folds a **clean fan-out** — the suspension's eager `step_created` and `wait_created` writes — into `createBatch` calls of at most 32 events (mirroring the server's transaction budgets). The fold only engages when the World implements `createBatch`, the run is on slot identity, and the suspension carries no attribute writes, no hook writes, and no resilient step dispatch; everything else keeps the single-event path byte-for-byte. Lazy-inline steps keep deferring their `step_created` to the lazy start exactly as before. + +Per-event `409`s are tolerated the same way the single path tolerates `EntityConflictError` (a concurrent delivery already created the entity); any other per-event failure fails the suspension write the way a single-path rejection would. + +**Off by default** during burn-in; the variable is retained as the kill switch once it defaults on. + +## Follow-up + +The deferred sequential transition — holding `step_completed(N)` across the replay turn and committing `[step_completed(N), step_created(N+1), step_started(N+1)]` as one batch at the next lazy start — builds on this contract and ships separately. diff --git a/docs/content/docs/v5/changelog/meta.json b/docs/content/docs/v5/changelog/meta.json index d193d58e0b..63e53a02e7 100644 --- a/docs/content/docs/v5/changelog/meta.json +++ b/docs/content/docs/v5/changelog/meta.json @@ -7,7 +7,8 @@ "resilient-start", "lazy-event-creation", "turbo-mode", - "step-message-ownership" + "step-message-ownership", + "batched-event-writes" ], "defaultOpen": false } diff --git a/docs/content/docs/v5/configuration/worlds.mdx b/docs/content/docs/v5/configuration/worlds.mdx index 72487e4a1f..135fbe4a52 100644 --- a/docs/content/docs/v5/configuration/worlds.mdx +++ b/docs/content/docs/v5/configuration/worlds.mdx @@ -274,6 +274,14 @@ Platform-provided values such as `VERCEL_DEPLOYMENT_ID`, `VERCEL_PROJECT_ID`, an - Default: `1000` - Maximum stream chunks written in one Vercel World request. Larger batches are split. +### `WORKFLOW_BATCH_TRANSITIONS` + +- Surface: environment variable +- Default: unset (off) +- Set to `1` to fold a suspension's eager `step_created` and `wait_created` writes into batched `events.createBatch` calls (one durable write with per-event outcomes) on Worlds that implement the optional batch API. + +Only engages when the World implements `events.createBatch` (the Vercel World does; Local and Postgres do not), the run's spec version supports slot identity (≥ 6), and the suspension carries no attribute writes, hook writes, or resilient step dispatch — everything else keeps the single-event path unchanged. Batches are capped at 32 events; larger fan-outs commit in successive batches. See the [batched event writes changelog](/docs/changelog/batched-event-writes) for the World API contract. + ### `WORKFLOW_EVENTS_TRANSPORT` - Factory option: none diff --git a/packages/core/src/runtime/constants.ts b/packages/core/src/runtime/constants.ts index f04647d10b..6bd23fd6d0 100644 --- a/packages/core/src/runtime/constants.ts +++ b/packages/core/src/runtime/constants.ts @@ -254,6 +254,37 @@ export function isResilientStepDispatchEnabled(): boolean { return process.env.WORKFLOW_RESILIENT_STEP_DISPATCH === '1'; } +/** + * Whether batched event transitions are enabled: the suspension handler folds + * a clean fan-out's `step_created` + `wait_created` writes into one + * `world.events.createBatch` call (one durable write, per-event outcomes) + * instead of one write per event. Only engages when the World implements the + * optional `events.createBatch` AND the run is on slot identity + * (specVersion >= 6) AND the suspension carries no attribute/hook writes and + * no resilient step dispatch — everything else keeps the single-event path + * byte-for-byte. + * + * **Off by default.** Enable via `WORKFLOW_BATCH_TRANSITIONS=1`. Staged like + * resilient step dispatch: opt-in for burn-in against the batch-tagged + * slot-conflict and throttle metrics, then default-on with this variable + * retained as the kill switch. + */ +export function isBatchTransitionsEnabled(): boolean { + return process.env.WORKFLOW_BATCH_TRANSITIONS === '1'; +} + +/** + * Ceiling on events per `createBatch` call from the batched fan-out fold. + * Mirrors the server's transaction budgets with a comfortable margin: each + * fan-out event costs 2 transaction items server-side (entity + event row) + * against the 100-item DynamoDB cap, and inline payloads count against a + * 768 KB byte budget — 32 events stays well under both, and a fan-out larger + * than this simply commits in successive batches (split batches lose + * cross-batch atomicity, which is exactly today's per-event-write crash + * surface — every batch still converges on retry via per-event 409s). + */ +export const MAX_BATCH_FANOUT_EVENTS = 32; + const warnedMaxEventsValues = new Set(); /** diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index d34814500b..d268c538bc 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -912,3 +912,277 @@ describe('retainedStepInputsSafe (serialization passivity gate)', () => { expect(result.lazyInlineSteps).toHaveLength(1); }); }); + +describe('handleSuspension batched fan-out', () => { + const slotRun: WorkflowRun = { ...run, specVersion: 6 }; + + function createBatchWorld( + eventsCreate: ReturnType, + createBatch?: ReturnType + ): World { + return { + events: { + create: eventsCreate, + ...(createBatch ? { createBatch } : {}), + }, + getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), + } as unknown as World; + } + + /** createBatch mock answering every event with a 200 at consecutive slots. */ + function successfulCreateBatch(firstSlot = 10) { + let slot = firstSlot; + return vi.fn().mockImplementation(async (_runId, events) => ({ + results: events.map(({ event }: { event: { eventType: string } }) => ({ + status: 200, + event: { ...event, eventId: slotToEventId(slot++) }, + })), + })); + } + + function stepsAndWait(stepIds: string[], waitId?: string) { + const pending = new Map( + stepIds.map((id) => [ + id, + { type: 'step' as const, correlationId: id, stepName: id, args: [] }, + ]) + ); + if (waitId) { + pending.set(waitId, { + type: 'wait' as const, + correlationId: waitId, + resumeAt: new Date(Date.now() + 60_000), + }); + } + return pending as ConstructorParameters[0]; + } + + beforeEach(() => { + vi.stubEnv('WORKFLOW_BATCH_TRANSITIONS', '1'); + // Cap lazy-inline deferral at 1 so only the first step defers its + // step_created and the rest take the eager path where the fold engages; + // the cap interaction has its own test below. + vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '1'); + }); + + afterEach(() => { + vi.unstubAllEnvs(); + }); + + it('folds eager step and wait creates into one createBatch, in order', async () => { + const eventsCreate = vi.fn(); + const createBatch = successfulCreateBatch(); + const world = createBatchWorld(eventsCreate, createBatch); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension( + stepsAndWait(['s1', 's2', 's3'], 'wait_1'), + globalThis + ), + world, + run: slotRun, + }); + + expect(createBatch).toHaveBeenCalledTimes(1); + const [runId, events] = createBatch.mock.calls[0]; + expect(runId).toBe(slotRun.runId); + // s1 is lazy-inline deferred (cap 1); s2/s3 eager-create via the fold, + // then the wait — scheduling order preserved. + expect( + events.map((e: { event: { eventType: string } }) => e.event.eventType) + ).toEqual(['step_created', 'step_created', 'wait_created']); + expect( + events.map( + (e: { event: { correlationId: string } }) => e.event.correlationId + ) + ).toEqual(['s2', 's3', 'wait_1']); + // No single-event writes for the folded events. + expect(eventsCreate).not.toHaveBeenCalled(); + expect([...result.createdStepCorrelationIds].sort()).toEqual(['s2', 's3']); + }); + + it('tolerates a per-event 409 exactly like a single-path conflict', async () => { + const createBatch = vi.fn().mockImplementation(async (_runId, events) => ({ + results: events.map( + ( + { event }: { event: { eventType: string; correlationId: string } }, + index: number + ) => + index === 0 + ? { + status: 409, + error: 'conflict', + message: 'already created by an earlier delivery', + } + : { status: 200, event: { ...event, eventId: slotToEventId(11) } } + ), + })); + const world = createBatchWorld(vi.fn(), createBatch); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension( + stepsAndWait(['s1', 's2', 's3']), + globalThis + ), + world, + run: slotRun, + }); + + // s1 defers; of the folded pair, the conflicted step (s2) is not owned + // and the survivor (s3) is. + expect([...result.createdStepCorrelationIds]).toEqual(['s3']); + }); + + it('fails the suspension on a non-409 per-event failure', async () => { + const createBatch = vi.fn().mockImplementation(async (_runId, events) => ({ + results: events.map(() => ({ + status: 410, + error: 'gone', + message: 'run already finished', + })), + })); + const world = createBatchWorld(vi.fn(), createBatch); + + await expect( + handleSuspension({ + suspension: new WorkflowSuspension( + stepsAndWait(['s1', 's2']), + globalThis + ), + world, + run: slotRun, + }) + ).rejects.toMatchObject({ status: 410 }); + }); + + it('keeps the single path when the flag is off', async () => { + vi.stubEnv('WORKFLOW_BATCH_TRANSITIONS', '0'); + const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => ({ + event, + })); + const createBatch = successfulCreateBatch(); + const world = createBatchWorld(eventsCreate, createBatch); + + await handleSuspension({ + suspension: new WorkflowSuspension( + stepsAndWait(['s1', 's2'], 'w1'), + globalThis + ), + world, + run: slotRun, + }); + + expect(createBatch).not.toHaveBeenCalled(); + // s1 defers; s2's eager create + the wait go out as single writes. + expect(eventsCreate).toHaveBeenCalledTimes(2); + }); + + it('keeps the single path when the World lacks createBatch', async () => { + const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => ({ + event, + })); + const world = createBatchWorld(eventsCreate); + + await handleSuspension({ + suspension: new WorkflowSuspension( + stepsAndWait(['s1', 's2']), + globalThis + ), + world, + run: slotRun, + }); + + expect(eventsCreate).toHaveBeenCalledTimes(1); + }); + + it('keeps the single path on a pre-slot-identity run', async () => { + const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => ({ + event, + })); + const createBatch = successfulCreateBatch(); + const world = createBatchWorld(eventsCreate, createBatch); + + await handleSuspension({ + suspension: new WorkflowSuspension( + stepsAndWait(['s1', 's2']), + globalThis + ), + world, + run: { ...run, specVersion: 5 }, + }); + + expect(createBatch).not.toHaveBeenCalled(); + expect(eventsCreate).toHaveBeenCalledTimes(1); + }); + + it('keeps the single path when the suspension carries hook writes', async () => { + const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => ({ + event: { ...event, eventType: event.eventType }, + hook: { hookId: 'hook_1', token: 'tok' }, + })); + const createBatch = successfulCreateBatch(); + const world = createBatchWorld(eventsCreate, createBatch); + const pending = stepsAndWait(['s1']) as Map; + pending.set('hook_1', { + type: 'hook' as const, + correlationId: 'hook_1', + token: 'order:456', + }); + + await handleSuspension({ + suspension: new WorkflowSuspension( + pending as ConstructorParameters[0], + globalThis + ), + world, + run: slotRun, + }); + + expect(createBatch).not.toHaveBeenCalled(); + }); + + it('chunks a fan-out past MAX_BATCH_FANOUT_EVENTS', async () => { + const createBatch = successfulCreateBatch(); + const world = createBatchWorld(vi.fn(), createBatch); + const stepIds = Array.from({ length: 34 }, (_, i) => `s${i + 1}`); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension(stepsAndWait(stepIds), globalThis), + world, + run: slotRun, + }); + + // s1 defers; the remaining 33 eager creates chunk as 32 + 1. + expect(createBatch).toHaveBeenCalledTimes(2); + expect(createBatch.mock.calls[0][1]).toHaveLength(32); + expect(createBatch.mock.calls[1][1]).toHaveLength(1); + expect(result.createdStepCorrelationIds.size).toBe(33); + }); + + it('leaves lazy-inline deferred steps out of the batch', async () => { + // Default inline cap (3): s1..s3 defer their step_created for the lazy + // start; only s4 eager-creates, so the batch carries exactly one event. + vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '3'); + const createBatch = successfulCreateBatch(); + const world = createBatchWorld(vi.fn(), createBatch); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension( + stepsAndWait(['s1', 's2', 's3', 's4']), + globalThis + ), + world, + run: slotRun, + }); + + expect(result.lazyInlineSteps.map((s) => s.correlationId)).toEqual([ + 's1', + 's2', + 's3', + ]); + expect(createBatch).toHaveBeenCalledTimes(1); + expect(createBatch.mock.calls[0][1]).toHaveLength(1); + expect(createBatch.mock.calls[0][1][0].event.correlationId).toBe('s4'); + expect([...result.createdStepCorrelationIds]).toEqual(['s4']); + }); +}); diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index 0c303ad760..711de0c534 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -16,6 +16,7 @@ import { SPEC_VERSION_CURRENT, SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT, SPEC_VERSION_SUPPORTS_COMPRESSION, + SPEC_VERSION_SUPPORTS_SLOT_IDENTITY, type TraceCarrier, type ValidQueueName, type WorkflowRun, @@ -37,13 +38,16 @@ import * as Attribute from '../telemetry/semantic-conventions.js'; import { getAbortStreamIdFromToken } from '../util.js'; import { getMaxInlineSteps, + isBatchTransitionsEnabled, isResilientStepDispatchEnabled, + MAX_BATCH_FANOUT_EVENTS, MAX_RESILIENT_STEP_INPUT_BYTES, } from './constants.js'; import { absorbSkippedSlotReport, type EventCreator, type LoadedEventLog, + maxEventSlot, queueMessage, slotSnapshotParams, stepDispatchIdempotencyKey, @@ -697,6 +701,35 @@ export async function handleSuspension({ isResilientStepDispatchEnabled() && (run.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_CBOR_QUEUE_TRANSPORT; + // Batched fan-out: fold this suspension's step_created + wait_created + // writes into one `events.createBatch` call (one durable write, per-event + // outcomes) instead of one write per event. Engages only for a CLEAN + // fan-out — no attribute writes, no hook writes, no resilient dispatch + // (whose creates are each paired with a queue publish) — on a World that + // implements the optional method and a run whose events are slot-numbered. + // Everything outside the gate keeps the single-event path byte-for-byte. + const batchFanoutEligible = + isBatchTransitionsEnabled() && + typeof world.events.createBatch === 'function' && + (run.specVersion ?? 0) >= SPEC_VERSION_SUPPORTS_SLOT_IDENTITY && + !resilientDispatchEligible && + allHookItems.length === 0 && + attributeItems.length === 0; + /** + * The fold's collection, in scheduling order (steps in stepItems order, + * then waits). Step entries are enqueued by their prep promises (input + * dehydration runs concurrently, so entries are ordered by `order`, not by + * completion); the single flush op below awaits every prep, sorts, and + * commits the whole set through `createBatch`. + */ + const batchQueue: { + order: number; + kind: 'step' | 'wait'; + correlationId: string; + event: CreateEventRequest; + }[] = []; + const batchPreps: Promise[] = []; + // The trace carrier for resilient step dispatches, resolved at most once per // suspension (the per-step ops run concurrently and share it). let stepDispatchTraceCarrier: Promise | undefined; @@ -713,174 +746,209 @@ export async function handleSuspension({ // to caller — EXCEPT on the resilient dispatch path, which parallelizes the // create with the step's queue publish and reports it in // `queuedStepCorrelationIds`). + let batchOrderCounter = 0; for (const queueItem of stepItems) { if (stepsNeedingCreation.has(queueItem.correlationId)) { - ops.push( - (async () => { - // Per-step sink, merged below: the dehydrate wrapper emits span - // attributes from the sink it is handed, so sharing one across - // steps would re-emit (and misattribute) earlier steps' entries. - const stepGuestCode: GuestCodeStats = { executions: [] }; - const dehydratedInput = await dehydrateStepArguments( - { - args: queueItem.args, - closureVars: queueItem.closureVars, - thisVal: queueItem.thisVal, - }, - runId, - encryptionKey, - suspension.globalThis, - false, - compression, - stepGuestCode - ); - guestCodeStats.executions.push(...stepGuestCode.executions); - // Deferred (lazy) inline step: skip the step_created write — the - // caller's inline executeStep will send a lazy step_started carrying - // this input, and the world creates the step (entity + synthetic - // step_created event) atomically. We do NOT add it to - // createdStepCorrelationIds; ownership is decided by that lazy - // step_started's atomic create-claim instead. - if (lazyInlineCorrelationIds.has(queueItem.correlationId)) { - lazyInlineByCorrelationId.set(queueItem.correlationId, { - correlationId: queueItem.correlationId, - stepName: queueItem.stepName, - dehydratedInput: dehydratedInput as SerializedData, - }); - return; - } - const stepEvent: CreateEventRequest = { - eventType: 'step_created' as const, - specVersion: SPEC_VERSION_CURRENT, + // Deterministic position in the batched fold (assigned in stepItems + // order, before the concurrent dehydration runs). + const stepOrder = batchOrderCounter++; + const stepOp = (async () => { + // Per-step sink, merged below: the dehydrate wrapper emits span + // attributes from the sink it is handed, so sharing one across + // steps would re-emit (and misattribute) earlier steps' entries. + const stepGuestCode: GuestCodeStats = { executions: [] }; + const dehydratedInput = await dehydrateStepArguments( + { + args: queueItem.args, + closureVars: queueItem.closureVars, + thisVal: queueItem.thisVal, + }, + runId, + encryptionKey, + suspension.globalThis, + false, + compression, + stepGuestCode + ); + guestCodeStats.executions.push(...stepGuestCode.executions); + // Deferred (lazy) inline step: skip the step_created write — the + // caller's inline executeStep will send a lazy step_started carrying + // this input, and the world creates the step (entity + synthetic + // step_created event) atomically. We do NOT add it to + // createdStepCorrelationIds; ownership is decided by that lazy + // step_started's atomic create-claim instead. + if (lazyInlineCorrelationIds.has(queueItem.correlationId)) { + lazyInlineByCorrelationId.set(queueItem.correlationId, { correlationId: queueItem.correlationId, - eventData: { - stepName: queueItem.stepName, - workflowName: run.workflowName, - input: dehydratedInput as SerializedData, - }, - }; + stepName: queueItem.stepName, + dehydratedInput: dehydratedInput as SerializedData, + }); + return; + } + const stepEvent: CreateEventRequest = { + eventType: 'step_created' as const, + specVersion: SPEC_VERSION_CURRENT, + correlationId: queueItem.correlationId, + eventData: { + stepName: queueItem.stepName, + workflowName: run.workflowName, + input: dehydratedInput as SerializedData, + }, + }; - // Resilient step dispatch: fire the step_created write and the - // step-execution queue publish in parallel — the message carries the - // same serialized input (`stepInput`) so the consumer can - // idempotently re-ensure the event if the direct write failed - // transiently. Mirrors the resilient start (`runInput`) and - // resilient hook resume (`hookInput`) patterns. Only for inputs the - // queue message can safely carry (binary, under the VQS size cap). - if ( - resilientDispatchEligible && - dehydratedInput instanceof Uint8Array && - dehydratedInput.byteLength <= MAX_RESILIENT_STEP_INPUT_BYTES - ) { - await ensureRunReady(); - const traceCarrier = await getStepDispatchTraceCarrier(); - const [createResult, queueResult] = await Promise.allSettled([ - createGuarded(stepEvent, { requestId }), - queueMessage( - world, - // biome-ignore lint/style/noNonNullAssertion: implied by resilientDispatchEligible - stepDispatch!.queueName, - { - runId, - stepId: queueItem.correlationId, - stepName: queueItem.stepName, - traceCarrier, - requestedAt: new Date(), - stepInput: { input: dehydratedInput }, - }, - // Same key as the caller's dispatch pass and any concurrent - // handler's — redundant publishes for this step dedupe. The - // key is step-identity-scoped so a revoked message for a - // reassigned correlation id cannot absorb the corrected - // schedule's dispatch — see stepDispatchIdempotencyKey. - { - idempotencyKey: stepDispatchIdempotencyKey( - queueItem.correlationId, - queueItem.stepName - ), - } - ), - ]); - // Queue failure is always fatal for this suspension pass: without - // the message the step would rely on the create alone, and if the - // create ALSO failed there would be no durable record at all. - // Propagating redelivers the orchestrator message, which - // re-creates the (idempotent) step_created and re-dispatches — - // the same recovery as the sequential path. - if (queueResult.status === 'rejected') { - throw queueResult.reason; - } - queuedStepCorrelationIds.add(queueItem.correlationId); - if (createResult.status === 'rejected') { - const err = createResult.reason; - if (EntityConflictError.is(err)) { - // Concurrent handler wrote it first — same as the sequential - // path. The step message is already out; a duplicate publish - // by that handler dedupes on the shared idempotency key. - runtimeLogger.info('Step already exists, continuing', { - workflowRunId: runId, - correlationId: queueItem.correlationId, - message: err.message, - }); - } else if (isRetryableWorldError(err)) { - // Resilient: the write failed transiently (429 / 5xx / - // transport) but the step message — carrying the same - // serialized input — was published, so the consumer - // idempotently re-ensures the step_created before executing. - resilientDispatchRecovered++; - runtimeLogger.warn( - 'Step creation event write failed, but the step was ' + - 'dispatched via the queue. The step_created event will ' + - 'be ensured by the queue consumer.', - { - workflowRunId: runId, - correlationId: queueItem.correlationId, - stepName: queueItem.stepName, - error: err instanceof Error ? err.message : String(err), - } - ); - } else { - throw err; + // Resilient step dispatch: fire the step_created write and the + // step-execution queue publish in parallel — the message carries the + // same serialized input (`stepInput`) so the consumer can + // idempotently re-ensure the event if the direct write failed + // transiently. Mirrors the resilient start (`runInput`) and + // resilient hook resume (`hookInput`) patterns. Only for inputs the + // queue message can safely carry (binary, under the VQS size cap). + if ( + resilientDispatchEligible && + dehydratedInput instanceof Uint8Array && + dehydratedInput.byteLength <= MAX_RESILIENT_STEP_INPUT_BYTES + ) { + await ensureRunReady(); + const traceCarrier = await getStepDispatchTraceCarrier(); + const [createResult, queueResult] = await Promise.allSettled([ + createGuarded(stepEvent, { requestId }), + queueMessage( + world, + // biome-ignore lint/style/noNonNullAssertion: implied by resilientDispatchEligible + stepDispatch!.queueName, + { + runId, + stepId: queueItem.correlationId, + stepName: queueItem.stepName, + traceCarrier, + requestedAt: new Date(), + stepInput: { input: dehydratedInput }, + }, + // Same key as the caller's dispatch pass and any concurrent + // handler's — redundant publishes for this step dedupe. The + // key is step-identity-scoped so a revoked message for a + // reassigned correlation id cannot absorb the corrected + // schedule's dispatch — see stepDispatchIdempotencyKey. + { + idempotencyKey: stepDispatchIdempotencyKey( + queueItem.correlationId, + queueItem.stepName + ), } - } else { - createdStepCorrelationIds.add(queueItem.correlationId); - } - return; + ), + ]); + // Queue failure is always fatal for this suspension pass: without + // the message the step would rely on the create alone, and if the + // create ALSO failed there would be no durable record at all. + // Propagating redelivers the orchestrator message, which + // re-creates the (idempotent) step_created and re-dispatches — + // the same recovery as the sequential path. + if (queueResult.status === 'rejected') { + throw queueResult.reason; } - - try { - await ensureRunReady(); - await createGuarded(stepEvent, { requestId }); - createdStepCorrelationIds.add(queueItem.correlationId); - } catch (err) { + queuedStepCorrelationIds.add(queueItem.correlationId); + if (createResult.status === 'rejected') { + const err = createResult.reason; if (EntityConflictError.is(err)) { + // Concurrent handler wrote it first — same as the sequential + // path. The step message is already out; a duplicate publish + // by that handler dedupes on the shared idempotency key. runtimeLogger.info('Step already exists, continuing', { workflowRunId: runId, correlationId: queueItem.correlationId, message: err.message, }); + } else if (isRetryableWorldError(err)) { + // Resilient: the write failed transiently (429 / 5xx / + // transport) but the step message — carrying the same + // serialized input — was published, so the consumer + // idempotently re-ensures the step_created before executing. + resilientDispatchRecovered++; + runtimeLogger.warn( + 'Step creation event write failed, but the step was ' + + 'dispatched via the queue. The step_created event will ' + + 'be ensured by the queue consumer.', + { + workflowRunId: runId, + correlationId: queueItem.correlationId, + stepName: queueItem.stepName, + error: err instanceof Error ? err.message : String(err), + } + ); } else { throw err; } + } else { + createdStepCorrelationIds.add(queueItem.correlationId); } - })() - ); + return; + } + + if (batchFanoutEligible) { + // Fold into the batch instead of writing here. The enclosing + // promise joins `batchPreps` (see the loop below), so the flush + // op cannot run before this step's input finished dehydrating. + batchQueue.push({ + order: stepOrder, + kind: 'step', + correlationId: queueItem.correlationId, + event: stepEvent, + }); + return; + } + + try { + await ensureRunReady(); + await createGuarded(stepEvent, { requestId }); + createdStepCorrelationIds.add(queueItem.correlationId); + } catch (err) { + if (EntityConflictError.is(err)) { + runtimeLogger.info('Step already exists, continuing', { + workflowRunId: runId, + correlationId: queueItem.correlationId, + message: err.message, + }); + } else { + throw err; + } + } + })(); + ops.push(stepOp); + if (batchFanoutEligible) { + // The flush op waits for every prep before committing; a prep that + // rejected already surfaces through `ops`, so the flush's own wait + // swallows it and commits whatever was successfully enqueued — + // preserving today's per-op independence. + batchPreps.push(stepOp.catch(() => {})); + } } } // Create wait events (same as V1) for (const queueItem of waitItems) { if (!queueItem.hasCreatedEvent) { + const waitEvent: CreateEventRequest = { + eventType: 'wait_created' as const, + specVersion: SPEC_VERSION_CURRENT, + correlationId: queueItem.correlationId, + eventData: { + resumeAt: queueItem.resumeAt, + }, + }; + if (batchFanoutEligible) { + // Waits need no dehydration, so they enqueue synchronously — after + // every step's order slot, preserving steps-then-waits scheduling + // order in the log. + batchQueue.push({ + order: batchOrderCounter++, + kind: 'wait', + correlationId: queueItem.correlationId, + event: waitEvent, + }); + continue; + } ops.push( (async () => { - const waitEvent: CreateEventRequest = { - eventType: 'wait_created' as const, - specVersion: SPEC_VERSION_CURRENT, - correlationId: queueItem.correlationId, - eventData: { - resumeAt: queueItem.resumeAt, - }, - }; try { await ensureRunReady(); await createGuarded(waitEvent, { requestId }); @@ -900,6 +968,94 @@ export async function handleSuspension({ } } + // The batched fold's flush: ONE durable write for the whole clean fan-out + // (chunked at MAX_BATCH_FANOUT_EVENTS), joining `ops` like the per-event + // writes it replaces so settlePhase semantics are unchanged. Each event + // reports the outcome its own single create would have had: a 409 is the + // same already-exists tolerance as the single path, anything else fails + // the op the way a single-path rejection would. + if (batchFanoutEligible) { + ops.push( + (async () => { + // Preps that rejected already surface through their own `ops` + // entries; the fold commits whatever was successfully enqueued, + // preserving today's per-op independence. + await Promise.all(batchPreps); + if (batchQueue.length === 0) { + return; + } + const entries = [...batchQueue].sort((a, b) => a.order - b.order); + await ensureRunReady(); + for ( + let start = 0; + start < entries.length; + start += MAX_BATCH_FANOUT_EVENTS + ) { + const chunk = entries.slice(start, start + MAX_BATCH_FANOUT_EVENTS); + const expectedFirstSlot = eventLog + ? (maxEventSlot(eventLog.events) ?? 0) + 1 + : undefined; + // biome-ignore lint/style/noNonNullAssertion: batchFanoutEligible implies presence + const { results } = await world.events.createBatch!( + runId, + chunk.map((entry) => ({ event: entry.event })) + ); + for (const [index, item] of results.entries()) { + const entry = chunk[index]; + if (item.error === undefined) { + if (entry.kind === 'step') { + createdStepCorrelationIds.add(entry.correlationId); + } + continue; + } + if (item.status === 409) { + // Same tolerance as the single path's EntityConflictError: a + // concurrent or earlier delivery already created it. + runtimeLogger.info( + entry.kind === 'step' + ? 'Step already exists, continuing' + : 'Wait already exists, continuing', + { + workflowRunId: runId, + correlationId: entry.correlationId, + message: item.message, + } + ); + continue; + } + throw new WorkflowWorldError( + `batched ${entry.event.eventType} for ${entry.correlationId} ` + + `failed: ${item.error}: ${item.message}`, + { status: item.status } + ); + } + // Slot-bump visibility: the batch endpoint has no bump-and-report, + // so a foreign event landing between our snapshot and the commit + // pushes the whole batch to higher slots WITHOUT handing us the + // skipped events. That is the same accepted exposure as a dropped + // truncated report on the single path (absorbSkippedSlotReport + // drops those whole): the local log continues without the foreign + // events and the next reload sees them. Logged so a bump is + // diagnosable rather than silent. + const firstCommitted = results.find( + (item) => item.error === undefined + )?.event; + if (expectedFirstSlot !== undefined && firstCommitted) { + const firstSlot = maxEventSlot([firstCommitted]); + if (firstSlot !== undefined && firstSlot > expectedFirstSlot) { + runtimeLogger.debug('Batched fan-out committed above snapshot', { + workflowRunId: runId, + expectedFirstSlot, + firstSlot, + skipped: firstSlot - expectedFirstSlot, + }); + } + } + } + })() + ); + } + for (const queueItem of attributeItems) { ops.push( (async () => { From 8863390cbc282a44ef9c37ca9ac343693e601845 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 14 Aug 2026 14:00:43 -0700 Subject: [PATCH 3/6] docs: make the createBatch signature sample typecheck MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The docs code-sample checker compiles every ts block as standalone TypeScript. The signature block now imports the real types from @workflow/world and declares the method on an interface — so the doc is a live contract test against the actual spec — and the illustrative type-shape excerpts carry the @skip-typecheck marker with the canonical definitions referenced. Co-Authored-By: Claude Fable 5 --- .../v5/changelog/batched-event-writes.mdx | 23 +++++++++++++++---- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/docs/content/docs/v5/changelog/batched-event-writes.mdx b/docs/content/docs/v5/changelog/batched-event-writes.mdx index 6f526156c0..5b8e171647 100644 --- a/docs/content/docs/v5/changelog/batched-event-writes.mdx +++ b/docs/content/docs/v5/changelog/batched-event-writes.mdx @@ -14,12 +14,25 @@ A workflow suspension that schedules several steps and waits previously wrote on `Storage['events']` gains one **optional** method: ```ts -createBatch?( - runId: string, - events: BatchEventRequest[], - params?: CreateEventBatchParams -): Promise; +import type { + BatchEventRequest, + CreateEventBatchParams, + EventBatchResult, +} from '@workflow/world'; + +interface BatchCapableEvents { + createBatch?( + runId: string, + events: BatchEventRequest[], + params?: CreateEventBatchParams + ): Promise; +} +``` +The supporting types, excerpted (canonical definitions live in `@workflow/world`): + +{/* @skip-typecheck illustrative excerpts of the canonical @workflow/world types */} +```ts interface BatchEventRequest { /** The event — the same discriminated union the single `create` takes. */ event: CreateEventRequest; From a739eef54a932e89f814803fbe5982f874f6f892 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 14 Aug 2026 14:51:50 -0700 Subject: [PATCH 4/6] feat(core): batched event transitions default ON, WORKFLOW_BATCH_TRANSITIONS=0 escape hatch Flips isBatchTransitionsEnabled to the WORKFLOW_TURBO kill-switch shape: default ON, disabled only by an explicit '0'/'false' (case-insensitive). Disabling restores the exact prior one-write-per-event path. The suspension-handler batch tests now run with the env var UNSET, proving the default engages; the kill-switch test pins the escape hatch. Docs updated in both places: the worlds configuration reference entry documents default-on + the escape hatch, and the changelog entry's rollout note points at it. Core suite: 2157 passed with the default flipped. Co-Authored-By: Claude Fable 5 --- .../docs/v5/changelog/batched-event-writes.mdx | 4 ++-- docs/content/docs/v5/configuration/worlds.mdx | 6 +++--- packages/core/src/runtime/constants.ts | 12 +++++++----- packages/core/src/runtime/suspension-handler.test.ts | 6 ++++-- 4 files changed, 16 insertions(+), 12 deletions(-) diff --git a/docs/content/docs/v5/changelog/batched-event-writes.mdx b/docs/content/docs/v5/changelog/batched-event-writes.mdx index 5b8e171647..c1b7ddc054 100644 --- a/docs/content/docs/v5/changelog/batched-event-writes.mdx +++ b/docs/content/docs/v5/changelog/batched-event-writes.mdx @@ -60,11 +60,11 @@ The contract: ## The runtime integration (suspension fan-out fold) -With `WORKFLOW_BATCH_TRANSITIONS=1`, the suspension handler folds a **clean fan-out** — the suspension's eager `step_created` and `wait_created` writes — into `createBatch` calls of at most 32 events (mirroring the server's transaction budgets). The fold only engages when the World implements `createBatch`, the run is on slot identity, and the suspension carries no attribute writes, no hook writes, and no resilient step dispatch; everything else keeps the single-event path byte-for-byte. Lazy-inline steps keep deferring their `step_created` to the lazy start exactly as before. +**On by default.** The suspension handler folds a **clean fan-out** — the suspension's eager `step_created` and `wait_created` writes — into `createBatch` calls of at most 32 events (mirroring the server's transaction budgets). The fold only engages when the World implements `createBatch`, the run is on slot identity, and the suspension carries no attribute writes, no hook writes, and no resilient step dispatch; everything else keeps the single-event path byte-for-byte. Lazy-inline steps keep deferring their `step_created` to the lazy start exactly as before. Per-event `409`s are tolerated the same way the single path tolerates `EntityConflictError` (a concurrent delivery already created the entity); any other per-event failure fails the suspension write the way a single-path rejection would. -**Off by default** during burn-in; the variable is retained as the kill switch once it defaults on. +**Escape hatch:** set `WORKFLOW_BATCH_TRANSITIONS=0` (or `false`) to disable batching and restore the exact prior one-write-per-event path — see [`WORKFLOW_BATCH_TRANSITIONS`](/docs/configuration/worlds#workflow_batch_transitions). ## Follow-up diff --git a/docs/content/docs/v5/configuration/worlds.mdx b/docs/content/docs/v5/configuration/worlds.mdx index 135fbe4a52..7a7a96effa 100644 --- a/docs/content/docs/v5/configuration/worlds.mdx +++ b/docs/content/docs/v5/configuration/worlds.mdx @@ -277,10 +277,10 @@ Platform-provided values such as `VERCEL_DEPLOYMENT_ID`, `VERCEL_PROJECT_ID`, an ### `WORKFLOW_BATCH_TRANSITIONS` - Surface: environment variable -- Default: unset (off) -- Set to `1` to fold a suspension's eager `step_created` and `wait_created` writes into batched `events.createBatch` calls (one durable write with per-event outcomes) on Worlds that implement the optional batch API. +- Default: on +- Set to `0` (or `false`) to **disable** batched event writes — the escape hatch that restores the exact prior one-write-per-event path. -Only engages when the World implements `events.createBatch` (the Vercel World does; Local and Postgres do not), the run's spec version supports slot identity (≥ 6), and the suspension carries no attribute writes, hook writes, or resilient step dispatch — everything else keeps the single-event path unchanged. Batches are capped at 32 events; larger fan-outs commit in successive batches. See the [batched event writes changelog](/docs/changelog/batched-event-writes) for the World API contract. +When enabled (the default), a suspension's eager `step_created` and `wait_created` writes fold into batched `events.createBatch` calls (one durable write with per-event outcomes) on Worlds that implement the optional batch API. The fold only engages when the World implements `events.createBatch` (the Vercel World does; Local and Postgres do not), the run's spec version supports slot identity (≥ 6), and the suspension carries no attribute writes, hook writes, or resilient step dispatch — everything else keeps the single-event path unchanged, so disabling is only needed as an operational escape hatch. Batches are capped at 32 events; larger fan-outs commit in successive batches. See the [batched event writes changelog](/docs/changelog/batched-event-writes) for the World API contract. ### `WORKFLOW_EVENTS_TRANSPORT` diff --git a/packages/core/src/runtime/constants.ts b/packages/core/src/runtime/constants.ts index 6bd23fd6d0..26ea76bfd0 100644 --- a/packages/core/src/runtime/constants.ts +++ b/packages/core/src/runtime/constants.ts @@ -264,13 +264,15 @@ export function isResilientStepDispatchEnabled(): boolean { * no resilient step dispatch — everything else keeps the single-event path * byte-for-byte. * - * **Off by default.** Enable via `WORKFLOW_BATCH_TRANSITIONS=1`. Staged like - * resilient step dispatch: opt-in for burn-in against the batch-tagged - * slot-conflict and throttle metrics, then default-on with this variable - * retained as the kill switch. + * Reads `process.env.WORKFLOW_BATCH_TRANSITIONS` lazily. Default **ON**; + * disabled only by an explicit `'0'` / `'false'` (case-insensitive) — the + * operator escape hatch that restores the exact prior one-write-per-event + * path, mirroring `WORKFLOW_TURBO`'s kill-switch shape. */ export function isBatchTransitionsEnabled(): boolean { - return process.env.WORKFLOW_BATCH_TRANSITIONS === '1'; + const raw = process.env.WORKFLOW_BATCH_TRANSITIONS; + if (raw === undefined || raw === '') return true; + return !(raw === '0' || raw.toLowerCase() === 'false'); } /** diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index d268c538bc..ce5edb0589 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -958,7 +958,9 @@ describe('handleSuspension batched fan-out', () => { } beforeEach(() => { - vi.stubEnv('WORKFLOW_BATCH_TRANSITIONS', '1'); + // No WORKFLOW_BATCH_TRANSITIONS stub: the fold is DEFAULT ON, so these + // tests exercising it with an unset env prove the default engages. The + // kill switch has its own test below. // Cap lazy-inline deferral at 1 so only the first step defers its // step_created and the rest take the eager path where the fold engages; // the cap interaction has its own test below. @@ -1055,7 +1057,7 @@ describe('handleSuspension batched fan-out', () => { ).rejects.toMatchObject({ status: 410 }); }); - it('keeps the single path when the flag is off', async () => { + it('keeps the single path when the kill switch disables batching', async () => { vi.stubEnv('WORKFLOW_BATCH_TRANSITIONS', '0'); const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => ({ event, From e10c234f9a8344987843d2ae6f2a6bf6b2e6ae01 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 14 Aug 2026 14:57:03 -0700 Subject: [PATCH 5/6] fix(core): route a lone eager event through the single path, never a batch of one A batch of one gains nothing over the single write (same round trip) and loses the slot-snapshot params and bump-and-report that createGuarded provides. The flush now takes the ordinary single path when exactly one eager event survived collection, with the same conflict tolerance and ownership bookkeeping. So the batch endpoint is used exactly when a suspension has TWO OR MORE batchable eager events; a lone create is wire-identical to the pre-batching behavior. Tests updated accordingly (a new single-entry-fallback test; the 410 and lazy-inline tests now use genuinely multi-event folds). Co-Authored-By: Claude Fable 5 --- .../src/runtime/suspension-handler.test.ts | 47 ++++++++++++++++--- .../core/src/runtime/suspension-handler.ts | 30 ++++++++++++ 2 files changed, 71 insertions(+), 6 deletions(-) diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index ce5edb0589..aee668d278 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -1048,7 +1048,8 @@ describe('handleSuspension batched fan-out', () => { await expect( handleSuspension({ suspension: new WorkflowSuspension( - stepsAndWait(['s1', 's2']), + // s1 defers; s2 + s3 form a real (multi-event) batch. + stepsAndWait(['s1', 's2', 's3']), globalThis ), world, @@ -1143,6 +1144,36 @@ describe('handleSuspension batched fan-out', () => { expect(createBatch).not.toHaveBeenCalled(); }); + it('routes a lone eager event through the single path, never a batch of one', async () => { + const eventsCreate = vi.fn().mockImplementation(async (_runId, event) => ({ + event, + })); + const createBatch = successfulCreateBatch(); + const world = createBatchWorld(eventsCreate, createBatch); + + // Two steps: s1 lazy-defers (cap 1), leaving exactly one eager create. + const result = await handleSuspension({ + suspension: new WorkflowSuspension( + stepsAndWait(['s1', 's2']), + globalThis + ), + world, + run: slotRun, + }); + + expect(createBatch).not.toHaveBeenCalled(); + expect(eventsCreate).toHaveBeenCalledTimes(1); + expect(eventsCreate).toHaveBeenCalledWith( + slotRun.runId, + expect.objectContaining({ + eventType: 'step_created', + correlationId: 's2', + }), + expect.anything() + ); + expect([...result.createdStepCorrelationIds]).toEqual(['s2']); + }); + it('chunks a fan-out past MAX_BATCH_FANOUT_EVENTS', async () => { const createBatch = successfulCreateBatch(); const world = createBatchWorld(vi.fn(), createBatch); @@ -1163,14 +1194,14 @@ describe('handleSuspension batched fan-out', () => { it('leaves lazy-inline deferred steps out of the batch', async () => { // Default inline cap (3): s1..s3 defer their step_created for the lazy - // start; only s4 eager-creates, so the batch carries exactly one event. + // start; s4 + s5 eager-create, so the batch carries exactly those two. vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '3'); const createBatch = successfulCreateBatch(); const world = createBatchWorld(vi.fn(), createBatch); const result = await handleSuspension({ suspension: new WorkflowSuspension( - stepsAndWait(['s1', 's2', 's3', 's4']), + stepsAndWait(['s1', 's2', 's3', 's4', 's5']), globalThis ), world, @@ -1183,8 +1214,12 @@ describe('handleSuspension batched fan-out', () => { 's3', ]); expect(createBatch).toHaveBeenCalledTimes(1); - expect(createBatch.mock.calls[0][1]).toHaveLength(1); - expect(createBatch.mock.calls[0][1][0].event.correlationId).toBe('s4'); - expect([...result.createdStepCorrelationIds]).toEqual(['s4']); + expect(createBatch.mock.calls[0][1]).toHaveLength(2); + expect( + createBatch.mock.calls[0][1].map( + (e: { event: { correlationId: string } }) => e.event.correlationId + ) + ).toEqual(['s4', 's5']); + expect([...result.createdStepCorrelationIds].sort()).toEqual(['s4', 's5']); }); }); diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index 711de0c534..ba57076e15 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -986,6 +986,36 @@ export async function handleSuspension({ } const entries = [...batchQueue].sort((a, b) => a.order - b.order); await ensureRunReady(); + // A batch of ONE gains nothing over the single write (same round + // trip) and loses the slot-snapshot params + bump-and-report that + // createGuarded provides — so a lone eager event takes the ordinary + // single path, with the same conflict tolerance and ownership + // bookkeeping it would have had without the fold. + if (entries.length === 1) { + const [entry] = entries; + try { + await createGuarded(entry.event, { requestId }); + if (entry.kind === 'step') { + createdStepCorrelationIds.add(entry.correlationId); + } + } catch (err) { + if (EntityConflictError.is(err)) { + runtimeLogger.info( + entry.kind === 'step' + ? 'Step already exists, continuing' + : 'Wait already exists, continuing', + { + workflowRunId: runId, + correlationId: entry.correlationId, + message: err.message, + } + ); + } else { + throw err; + } + } + return; + } for ( let start = 0; start < entries.length; From 6ec5a302ae792893fe65d97477facdf65e7e5af1 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 14 Aug 2026 16:49:38 -0700 Subject: [PATCH 6/6] Address review: changeset, retry-convergence gating, result hardening, attribution - Add the missing changeset (@workflow/world, @workflow/world-vercel, @workflow/core). - Scope the batch retry contract honestly: reject hook_received in a batch (no entity condition - a retry would append the delivery twice), derive batchIdempotent from the batch's contents (standalone bare step_started / step_retrying batches run single-attempt), and make an explicit verdict replace the per-type matrix instead of only widening it. Docs in the World spec, the wrapper, and the changelog now state which shapes converge. - Harden batch result decoding: success requires literal status 200, and a result whose committed event type differs from the frame submitted at that index is a protocol violation (SCHEMA_VALIDATION), not a success. - Thread per-write attribution: CreateEventBatchParams.requestId -> per-frame vercelId; the suspension fold passes its requestId like the single path. - Honor resolveData in world-vercel createBatch ('all' -> resolve). - Fix the multi-chunk slot-bump diagnostic: the expectation now advances past each chunk's own committed events, so chunk 2+ no longer misreads the fold's earlier chunks as foreign skips. - Document the actual slot rule on createBatch (no skipped-event report; the caller's view stays a strict prefix and the next reload self-corrects), Worlds' size limits, born-running pair payload placement, ordering barriers for non-batchable events, and the HTTP-only transport; tag batch spans with per-type shape and byte size. --- .changeset/batched-event-writes.md | 7 + .../v5/changelog/batched-event-writes.mdx | 4 +- .../core/src/runtime/suspension-handler.ts | 29 ++- packages/world-vercel/src/event-retry.ts | 30 ++- .../world-vercel/src/events-batch.test.ts | 182 ++++++++++++++++++ packages/world-vercel/src/events-v4.ts | 50 ++++- packages/world-vercel/src/events.ts | 49 ++++- packages/world/src/events.ts | 32 ++- packages/world/src/interfaces.ts | 21 +- 9 files changed, 374 insertions(+), 30 deletions(-) create mode 100644 .changeset/batched-event-writes.md diff --git a/.changeset/batched-event-writes.md b/.changeset/batched-event-writes.md new file mode 100644 index 0000000000..834f55bd2c --- /dev/null +++ b/.changeset/batched-event-writes.md @@ -0,0 +1,7 @@ +--- +'@workflow/world': patch +'@workflow/world-vercel': patch +'@workflow/core': patch +--- + +Batched event writes: add the optional `events.createBatch` World API (ordered events, one durable write, per-event outcomes), implement it in `@workflow/world-vercel` against `POST /v4/runs/:runId/events/batch` (slot-identity runs only), and fold clean suspension fan-outs — eager `step_created` and `wait_created` writes — into batched writes in the runtime. On by default; disable with `WORKFLOW_BATCH_TRANSITIONS=0`. diff --git a/docs/content/docs/v5/changelog/batched-event-writes.mdx b/docs/content/docs/v5/changelog/batched-event-writes.mdx index c1b7ddc054..83ba431cc1 100644 --- a/docs/content/docs/v5/changelog/batched-event-writes.mdx +++ b/docs/content/docs/v5/changelog/batched-event-writes.mdx @@ -52,9 +52,9 @@ interface EventBatchResult { The contract: -- **Ordered**: events land in the run's log in request order (at consecutive slots, subject to bump-and-report semantics — a concurrent writer may push the whole batch to higher slots). +- **Ordered**: events land in the run's log in request order at consecutive slots. A concurrent writer may push the whole batch to slots above the caller's view; no skipped-event report accompanies the batch result, so a position-tracking caller compares committed slots against its expectation and reloads to observe what interleaved (its local view stays a strict prefix of the log — never a hole). - **Per-event outcomes**: the batch is processed as a whole, and each event reports what its own single `create` would have returned — `200` plus the materialized entity, or the single-path status/code (`409`/`conflict` for an event an earlier delivery already applied). Callers reuse their single-path conflict handling per event. -- **Idempotent on retry**: every batchable event is guarded by its own entity condition, so retrying a batch that (partially) committed converges to per-event `409`s with nothing written twice. +- **Idempotent on retry — for entity-conditioned shapes**: creates, terminal transitions, and the born-running pair are each guarded by their own entity condition, so retrying a batch of them that (partially) committed converges to per-event `409`s with nothing written twice. A standalone bare `step_started` or a `step_retrying` re-patches its step instead of converging, so `world-vercel` only auto-retries batches whose every event is retry-convergent (everything the runtime folds today is), and rejects `hook_received` in a batch outright. - **Method presence is the capability declaration.** A World that doesn't implement it keeps the single-event path; a World that implements it must make each attempt atomic (a lost race leaves nothing behind). `world-vercel` implements it against `POST /v4/runs/:runId/events/batch` (slot-identity runs only, i.e. specVersion ≥ 6). `world-local` and `world-postgres` deliberately do not — batching buys nothing for a local write. - **Not batchable** (Worlds reject the request): `run_created`, `run_started`, `run_cancelled`, `hook_created`, `hook_disposed`, `attr_set`, and multiple events targeting one entity — except `step_created` followed by `step_started` for the same step, which creates the step born-running. diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index ba57076e15..0e50ac4e09 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -1016,19 +1016,26 @@ export async function handleSuspension({ } return; } + // Expected next slot for the bump diagnostic below: seeded once from + // the caller's view and advanced past each chunk's own committed + // events, so chunk 2+ of a multi-chunk fan-out does not misread this + // fold's earlier chunks as foreign skips. + let expectedFirstSlot = eventLog + ? (maxEventSlot(eventLog.events) ?? 0) + 1 + : undefined; for ( let start = 0; start < entries.length; start += MAX_BATCH_FANOUT_EVENTS ) { const chunk = entries.slice(start, start + MAX_BATCH_FANOUT_EVENTS); - const expectedFirstSlot = eventLog - ? (maxEventSlot(eventLog.events) ?? 0) + 1 - : undefined; // biome-ignore lint/style/noNonNullAssertion: batchFanoutEligible implies presence const { results } = await world.events.createBatch!( runId, - chunk.map((entry) => ({ event: entry.event })) + chunk.map((entry) => ({ event: entry.event })), + // Per-write request attribution, same as the single path's + // createGuarded(…, { requestId }). + { requestId } ); for (const [index, item] of results.entries()) { const entry = chunk[index]; @@ -1081,6 +1088,20 @@ export async function handleSuspension({ }); } } + // Advance the expectation past this chunk's committed events so the + // next chunk's diagnostic measures only foreign interleaving. + const chunkMaxSlot = maxEventSlot( + results.flatMap((item) => + item.error === undefined && item.event ? [item.event] : [] + ) + ); + if ( + expectedFirstSlot !== undefined && + chunkMaxSlot !== undefined && + chunkMaxSlot >= expectedFirstSlot + ) { + expectedFirstSlot = chunkMaxSlot + 1; + } } })() ); diff --git a/packages/world-vercel/src/event-retry.ts b/packages/world-vercel/src/event-retry.ts index af8daa7b5a..3228bd45b8 100644 --- a/packages/world-vercel/src/event-retry.ts +++ b/packages/world-vercel/src/event-retry.ts @@ -332,15 +332,20 @@ export interface EventPostRetryOptions { */ idempotentHookResume?: boolean; /** - * Batch POST override: a `createBatch` request is idempotent-on-retry as a - * WHOLE regardless of the event types it carries, because every batchable - * event is guarded by its own entity condition — a retry of a batch that - * committed (or partially committed) converges to per-event 409 results - * with nothing written twice, which the caller already handles per event. - * The per-type eligibility matrix guards SINGLE posts, where e.g. a - * retried `step_started` would increment `attempt` unconditionally; in a - * batch that same start is fenced by the step create-claim, so the matrix - * does not apply. Definitive 4xx responses stay non-retryable regardless. + * Batch POST retry verdict, set by `createWorkflowRunEventBatch` from the + * batch's CONTENTS. `true` means every event in the batch converges on a + * retry of a committed attempt — entity-conditioned events (creates, + * terminal transitions) re-reject with 409, and a `step_started` is fenced + * by its born-running pair's create-claim — so the whole POST gets the + * standard transient retry policy and a re-send converges to per-event + * 409s with nothing written twice. `false` means at least one event does + * NOT converge (a standalone bare `step_started` re-increments `attempt`, + * a `step_retrying` re-patches its step), so the batch runs + * single-attempt and recovery is left to queue redelivery. When set (a + * batch call), this verdict REPLACES the per-type matrix entirely — the + * matrix classifies single posts and its entry for any one type says + * nothing about a mixed batch. Definitive 4xx responses stay + * non-retryable regardless. */ batchIdempotent?: boolean; } @@ -387,8 +392,13 @@ function isEligibleForTransientRetry( eventType: WorkflowEventType, options?: EventPostRetryOptions ): boolean { + // A batch call always carries an explicit verdict derived from every event + // it contains; the per-type matrix (keyed on the batch's FIRST event) must + // not override it in either direction. + if (options?.batchIdempotent !== undefined) { + return options.batchIdempotent; + } return ( - options?.batchIdempotent === true || (eventType === 'hook_received' && options?.idempotentHookResume === true) || (EVENT_RETRY_ELIGIBILITY[eventType]?.retryable ?? false) ); diff --git a/packages/world-vercel/src/events-batch.test.ts b/packages/world-vercel/src/events-batch.test.ts index 9b75af0179..b58e577dfb 100644 --- a/packages/world-vercel/src/events-batch.test.ts +++ b/packages/world-vercel/src/events-batch.test.ts @@ -383,3 +383,185 @@ describe('createWorkflowRunEventBatch', () => { ).rejects.toMatchObject({ status: 400 }); }); }); + +describe('createWorkflowRunEventBatch — retry-convergence and attribution', () => { + it('rejects hook_received without touching the network', async () => { + await expect( + createWorkflowRunEventBatch( + RUN_ID, + [ + { + event: { + eventType: 'hook_received', + specVersion: 6, + correlationId: 'hook_1', + eventData: { payload: utf8('"delivery"') }, + }, + }, + ], + undefined, + { token: 'test-token', dispatcher: mockAgent() } + ) + ).rejects.toMatchObject({ status: 400 }); + }); + + it('threads requestId to per-frame vercelId and honors resolveData', async () => { + const agent = mockAgent(); + let requestBody: Uint8Array | undefined; + agent + .get(ORIGIN) + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/batch`, + method: 'POST', + body: (raw) => { + requestBody = new Uint8Array(Buffer.from(raw, 'binary')); + return true; + }, + }) + .reply(200, fullSuccessBody(), { + headers: { 'content-type': 'application/cbor' }, + }); + + await createWorkflowRunEventBatch( + RUN_ID, + transitionEvents(), + { requestId: 'req_attrib_1', resolveData: 'all' }, + { token: 'test-token', dispatcher: agent } + ); + + // biome-ignore lint/style/noNonNullAssertion: interceptor ran + const frames = decodeBatchFrames(requestBody!); + for (const frame of frames) { + expect(frame.meta.vercelId).toBe('req_attrib_1'); + expect(frame.meta.remoteRefBehavior).toBe('resolve'); + } + agent.assertNoPendingInterceptors(); + }); + + it('does NOT retry a transient 5xx when the batch carries a bare step_started', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/batch`, + method: 'POST', + }) + .reply(503, JSON.stringify({ message: 'unavailable' }), { + headers: { 'content-type': 'application/json' }, + }); + + // started(step_b) does NOT follow its own step_created — a bare claim + // whose retry would re-increment attempt, so the POST is single-attempt. + const events: BatchEventRequest[] = [ + { + event: { + eventType: 'step_created', + specVersion: 6, + correlationId: 'step_a', + eventData: { + stepName: 'step-a', + workflowName: 'wf', + input: utf8('"a-input"'), + }, + }, + }, + { + event: { + eventType: 'step_started', + specVersion: 6, + correlationId: 'step_b', + eventData: { stepName: 'step-b' }, + }, + }, + ]; + + await expect( + createWorkflowRunEventBatch(RUN_ID, events, undefined, { + token: 'test-token', + dispatcher: agent, + }) + ).rejects.toMatchObject({ status: 503 }); + // The single interceptor was consumed exactly once — no in-process retry. + agent.assertNoPendingInterceptors(); + }); + + it('fails loudly when a success item is not literal status 200', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/batch`, + method: 'POST', + }) + .reply( + 200, + encode({ + results: [ + { status: 200, event: completedEvent, step: stepA }, + // Non-200 without error/message: neither a well-formed failure + // nor a success — must not be re-labeled a 200 by the body parse. + { status: 500, event: createdEvent, step: { ...stepB } }, + { status: 200, event: startedEvent, step: { ...stepB } }, + ], + }), + { headers: { 'content-type': 'application/cbor' } } + ); + + await expect( + createWorkflowRunEventBatch(RUN_ID, transitionEvents(), undefined, { + token: 'test-token', + dispatcher: agent, + }) + ).rejects.toMatchObject({ + code: 'SCHEMA_VALIDATION', + message: expect.stringContaining('index 1'), + }); + }); + + it('fails loudly when a result carries a different event type than submitted', async () => { + const agent = mockAgent(); + agent + .get(ORIGIN) + .intercept({ + path: `/api/v4/runs/${RUN_ID}/events/batch`, + method: 'POST', + }) + .reply( + 200, + encode({ + results: [ + { status: 200, event: completedEvent, step: stepA }, + // A wait_created event answering the step_created frame. + { + status: 200, + event: { + eventId: slotEventId(9), + runId: RUN_ID, + eventType: 'wait_created', + correlationId: 'step_b', + createdAt: CREATED_AT, + eventData: { resumeAt: CREATED_AT }, + }, + wait: { + runId: RUN_ID, + waitId: 'step_b', + status: 'pending', + resumeAt: CREATED_AT, + createdAt: CREATED_AT, + updatedAt: CREATED_AT, + }, + }, + { status: 200, event: startedEvent, step: { ...stepB } }, + ], + }), + { headers: { 'content-type': 'application/cbor' } } + ); + + await expect( + createWorkflowRunEventBatch(RUN_ID, transitionEvents(), undefined, { + token: 'test-token', + dispatcher: agent, + }) + ).rejects.toMatchObject({ code: 'SCHEMA_VALIDATION' }); + }); +}); diff --git a/packages/world-vercel/src/events-v4.ts b/packages/world-vercel/src/events-v4.ts index 220fcfc778..a3586f95de 100644 --- a/packages/world-vercel/src/events-v4.ts +++ b/packages/world-vercel/src/events-v4.ts @@ -841,9 +841,22 @@ const BatchItemFailureSchema = z.object({ * 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. Slot-identity runs only — an older server 404s the - * route and a pre-slot run is rejected with a 400, both of which callers - * treat as "fall back to single-event posts". + * frame in request order. + * + * 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, + * 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 + * for queue redelivery, whose replay re-derives an idempotent batch. + * + * HTTP-only by design: the WS event transport streams one frame per message + * and has no batch framing, so a WS-configured deployment still sends + * batches over HTTP (single-event writes keep their configured transport). */ export async function createWorkflowRunEventsBatchV4( input: CreateEventBatchV4Input, @@ -868,6 +881,13 @@ export async function createWorkflowRunEventsBatchV4( offset += frame.byteLength; } + // Per-type shape for the span, so mixed batches classify as what they + // carry rather than as their first event's type alone. + const typeCounts = new Map(); + for (const event of input.events) { + typeCounts.set(event.eventType, (typeCounts.get(event.eventType) ?? 0) + 1); + } + const url = `${baseUrl}/v4/runs/${encodeURIComponent(input.runId)}/events/batch`; const response = await fetchV4( url, @@ -878,6 +898,10 @@ export async function createWorkflowRunEventsBatchV4( ...WorkflowEventsTransport('http'), ...WorkflowEventType(input.events[0].eventType), 'workflow.batch.size': input.events.length, + 'workflow.batch.shape': [...typeCounts] + .map(([type, count]) => `${type}:${count}`) + .join(','), + 'workflow.batch.bytes': body.byteLength, } ); @@ -910,6 +934,16 @@ export async function createWorkflowRunEventsBatchV4( if (failure.success && failure.data.status !== 200) { return failure.data; } + // Success requires the LITERAL 200: an item with a non-200 status that + // failed the failure schema (e.g. missing error/message) must not fall + // through and be re-labeled a success by the body parse below. + if ((raw as { status?: unknown } | null)?.status !== 200) { + throw new WorkflowWorldError( + `v4 createEventBatch: result at index ${index} is neither a ` + + 'success (status 200) nor a well-formed failure', + { code: 'SCHEMA_VALIDATION' } + ); + } // Success items validate against the SAME per-type schema the single // POST uses, so a batched write and its single-path twin return // byte-equivalent bodies to the caller. @@ -921,6 +955,16 @@ export async function createWorkflowRunEventsBatchV4( { code: 'SCHEMA_VALIDATION', cause: parsed.error } ); } + // 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) { + throw new WorkflowWorldError( + `v4 createEventBatch: result at index ${index} carries a ` + + `${parsed.data.event.eventType} event, expected ${eventType}`, + { code: 'SCHEMA_VALIDATION' } + ); + } return { status: 200, ...parsed.data }; } ); diff --git a/packages/world-vercel/src/events.ts b/packages/world-vercel/src/events.ts index 79e959b5b8..1d0279229d 100644 --- a/packages/world-vercel/src/events.ts +++ b/packages/world-vercel/src/events.ts @@ -488,7 +488,7 @@ export async function getWorkflowRunEvents( export async function createWorkflowRunEventBatch( runId: string, events: BatchEventRequest[], - _params?: CreateEventBatchParams, + params?: CreateEventBatchParams, config?: APIConfig ): Promise { if (events.length === 0) { @@ -497,6 +497,20 @@ export async function createWorkflowRunEventBatch( { status: 400 } ); } + // Advisory `hook_received` has no entity condition: the server appends a + // fresh row on every attempt, so a retried batch would deliver the hook + // payload twice — and the atomic lazy-resume shape (the one dedupable + // form) is rejected by the batch route anyway. Nothing batches hook + // deliveries today; reject them here so the retry contract below stays + // honest instead of silently double-appending. + if (events.some(({ event }) => event.eventType === 'hook_received')) { + throw new WorkflowWorldError( + 'world-vercel: hook_received cannot be batched (it has no entity ' + + 'condition, so a batch retry would append the delivery twice); ' + + 'send it through the single-event path', + { status: 400 } + ); + } const inputs = events.map(({ event, occurredAt }) => { const { payload, meta } = splitEventDataForV4(event); return { @@ -508,17 +522,44 @@ export async function createWorkflowRunEventBatch( // the caller's logical time is what every replay observes. occurredAt: occurredAt ?? new Date(), // Batch responses carry entities for bookkeeping, not payload reads — - // the caller just produced every payload in this batch itself. - remoteRefBehavior: 'lazy' as const, + // default to lazy refs unless the caller explicitly asks for resolved + // data (the same `resolveData` mapping the read paths use). + remoteRefBehavior: + params?.resolveData === 'all' + ? ('resolve' as const) + : ('lazy' as const), + // Per-write request attribution, exactly like the single POST's + // `params.requestId` → `vercelId` threading — stamped per frame so + // batched usage facts carry the same attribution. + ...(params?.requestId ? { vercelId: params.requestId } : {}), payload, ...meta, }; }); + // In-process transient retry is safe only when EVERY event in the batch + // converges on a retry of a committed attempt: entity-conditioned events + // (creates, terminal transitions) re-reject with 409, and a step_started + // converges only as the second half of a born-running pair, where the + // pair's create-claim fences it. A standalone bare step_started re-patches + // a running step (attempt++) and step_retrying re-patches a pending one — + // a batch carrying either runs single-attempt and leaves transient-failure + // recovery to queue redelivery, exactly like their single POSTs. + const retryConvergent = events.every(({ event }, index) => { + if (event.eventType === 'step_started') { + const previous = events[index - 1]?.event; + return ( + previous?.eventType === 'step_created' && + previous.correlationId === event.correlationId + ); + } + return event.eventType !== 'step_retrying'; + }); + const wire = await withEventPostRetry( () => createWorkflowRunEventsBatchV4({ runId, events: inputs }, config), events[0].event.eventType, - { batchIdempotent: true } + { batchIdempotent: retryConvergent } ); return { diff --git a/packages/world/src/events.ts b/packages/world/src/events.ts index 08446b84f0..6f0c55ba6a 100644 --- a/packages/world/src/events.ts +++ b/packages/world/src/events.ts @@ -1036,6 +1036,13 @@ export interface BatchEventRequest { /** Per-batch parameters for {@link Storage.events.createBatch}. */ export interface CreateEventBatchParams { resolveData?: ResolveData; + /** + * Request id for per-write attribution, same as the single create's + * {@link CreateEventParams.requestId}: stamped on every event in the batch + * so a batched write's usage facts and telemetry carry the same request + * attribution its single-path twin would. + */ + requestId?: string; } /** @@ -1049,10 +1056,27 @@ export interface CreateEventBatchParams { * materialized entity the single create returns (`step` for step events, * `wait` for wait events, `run` for run terminals); * - rejection → the status code and error code the single create would have - * failed with (e.g. `409`/`conflict` for an event an earlier delivery - * already applied), so callers reuse their single-path conflict handling - * per event. A transport retry of a fully committed batch converges to - * all-409s with nothing written twice. + * failed with, so callers reuse their single-path conflict handling per + * event. A `409`/`conflict` means the entity was not in the prior state + * the event requires — most commonly because an earlier delivery already + * applied the same event, but possibly because the entity reached a + * DIFFERENT state (e.g. `step_completed` conflicting because the step + * failed). A 409 alone does not prove the equivalent effect was applied; + * a caller that needs effect-equivalence consults the entity (returned on + * sibling successes, or reloaded). + * + * The batch is atomic per attempt, not all-or-nothing across the submitted + * set: a World may drop rejected events and commit the survivors, so a batch + * can return a mix of 200s and 409s from one call. + * + * Retry semantics: a transport retry of a committed batch converges to + * per-event 409s ONLY for entity-conditioned events — creates, terminal + * transitions, and the born-running `step_created`+`step_started` pair + * (fenced by the pair's create). A standalone bare `step_started` or a + * `step_retrying` re-patches its step on every attempt and does NOT + * converge, and `hook_received` appends a new row per attempt — + * `world-vercel` rejects `hook_received` in a batch outright and only + * auto-retries batches whose every event is retry-convergent. */ export type BatchEventItemResult = | { diff --git a/packages/world/src/interfaces.ts b/packages/world/src/interfaces.ts index 037c2bc122..33542825f5 100644 --- a/packages/world/src/interfaces.ts +++ b/packages/world/src/interfaces.ts @@ -330,8 +330,13 @@ export interface Storage { * OPTIONAL batch write — append an ordered list of events to the run's * log in one durable, atomic-per-attempt write, with a per-event outcome * for each (see {@link BatchEventItemResult}). The events land in request - * order at consecutive slots (subject to bump-and-report semantics: a - * concurrent writer may push the whole batch to higher slots). + * order at consecutive slots. A concurrent writer may push the whole + * batch to slots above the caller's view of the log; no skipped-event + * report accompanies the result, so a position-tracking caller compares + * the committed slots against its expectation and reloads the log to + * observe what landed in between. Its local view stays a strict PREFIX + * of the log — never a hole — so replaying it stays correct and the + * next reload self-corrects. * * Presence of the method IS the capability declaration: the core runtime * batches only when the World implements it (and the run's spec version @@ -340,12 +345,22 @@ export interface Storage { * atomicity per attempt — a lost race must leave nothing behind — or not * implement it at all. * + * Size limits are the caller's problem: Worlds enforce their own caps + * (world-vercel enforces an event-count cap and a byte budget over frame + * meta plus inline-bound payloads) and reject an oversized batch with a + * request-level error. The core fold sizes its chunks accordingly. + * * Not expressible in a batch (Worlds reject the whole batch with a * request-level error): `run_created`, `run_started`, `run_cancelled`, * `hook_created`, `hook_disposed`, `attr_set`, and more events targeting * one entity than a single write can express (the one legal combination * is `step_created` followed by `step_started` for the same step, which - * creates the step born-running). + * creates the step born-running — the step's input MUST ride the + * `step_created`; a `step_started` carrying a payload rejects the whole + * batch). Events outside this list keep their own ordering requirements: + * a caller mixing a batch with single writes (hook or attribute events) + * owns those barriers itself — the core runtime simply never batches a + * suspension that carries hook or attribute writes. */ createBatch?( runId: string,