From a54893d5726ae42fbb404d3a6b0ebf2cc48d92d5 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 11 Sep 2026 01:21:55 -0700 Subject: [PATCH 1/2] perf(core): commit pre-claimed inline pairs in their own batch chunk The batched fan-out fold sorted the pre-claimed inline [step_created, step_started] pairs to the front and then filled the same chunk with plain step/wait creates up to MAX_BATCH_FANOUT_EVENTS. Only that chunk gates the inline bodies, so the bodies waited on a 32-row commit (a 61-item DynamoDB transaction on the Vercel backend) when all they needed was the pairs' own rows. Chunk the pair rows and the plain creates separately: the pairs get their own leading chunk(s), still adjacent and never split across a boundary, and the plain creates fill the subsequent chunks of 32, committing concurrently beside the pair chunk and gating only their own queue publishes. Eligibility, the batch-of-one rule, pairCommits gating, per-chunk publishes and the foreign-interleaving diagnostic are unchanged. Co-Authored-By: Claude Fable 5.1 --- .changeset/batch-pair-chunk.md | 5 + .../v5/changelog/batched-event-writes.mdx | 2 +- packages/core/src/runtime/constants.test.ts | 13 +- .../src/runtime/suspension-handler.test.ts | 223 +++++++++++++++--- .../core/src/runtime/suspension-handler.ts | 70 ++++-- 5 files changed, 248 insertions(+), 65 deletions(-) create mode 100644 .changeset/batch-pair-chunk.md diff --git a/.changeset/batch-pair-chunk.md b/.changeset/batch-pair-chunk.md new file mode 100644 index 0000000000..992c088571 --- /dev/null +++ b/.changeset/batch-pair-chunk.md @@ -0,0 +1,5 @@ +--- +"@workflow/core": patch +--- + +Commit the batched fan-out's pre-claimed inline `[step_created, step_started]` pairs in their own leading chunk, ahead of the plain step/wait creates, so the write that gates the inline bodies stays small and commits faster. diff --git a/docs/content/docs/v5/changelog/batched-event-writes.mdx b/docs/content/docs/v5/changelog/batched-event-writes.mdx index f2fd061426..5801e6d4b9 100644 --- a/docs/content/docs/v5/changelog/batched-event-writes.mdx +++ b/docs/content/docs/v5/changelog/batched-event-writes.mdx @@ -66,7 +66,7 @@ The contract: **Per-chunk continuation.** Each chunk's follow-on work starts the moment **that chunk** commits, not when the whole fold does: a chunk's step-execution queue messages publish right off its own commit (publish-after-create holds per step), and only the chunk carrying the inline pairs gates the replay's continuation: trailing chunks' commits and publishes are joined before the invocation can acknowledge its message, so the durability contract ("every create durable before ack") is unchanged. -**Pre-claimed inline pairs.** When the fold engages and has company for them (at least two inline steps, or one plus other batchable events), the steps the runtime is about to execute inline join the batch as adjacent `[step_created, step_started]` pairs: the created row carrying the input, the started row a bare ownership-stamped claim the World folds into a born-running create. The inline bodies start straight off the pair chunk's commit (in parallel with the queue publishes and any trailing chunks) with no per-step claim POST at all, and a pair that loses its atomic create-claim to a concurrent delivery skips its body exactly as a lost lazy claim does. A lone inline step with nothing else to batch keeps the optimistic lazy-start path, whose claim overlaps the body. +**Pre-claimed inline pairs.** When the fold engages and has company for them (at least two inline steps, or one plus other batchable events), the steps the runtime is about to execute inline join the batch as adjacent `[step_created, step_started]` pairs: the created row carrying the input, the started row a bare ownership-stamped claim the World folds into a born-running create. The pairs commit in a chunk of their own, ahead of the plain `step_created` and `wait_created` chunks, so the write the inline bodies wait for carries only two rows per inline step (a small transaction that commits faster than a full 32-event chunk) while the plain creates commit concurrently beside it. The inline bodies start straight off the pair chunk's commit (in parallel with the queue publishes and the sibling chunks) with no per-step claim POST at all, and a pair that loses its atomic create-claim to a concurrent delivery skips its body exactly as a lost lazy claim does. A lone inline step with nothing else to batch keeps the optimistic lazy-start path, whose claim overlaps the body. 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. A batch carrying a `step_started` (that is, any batch with inline pairs) is **not** retried in-process on a transport blip: a pair's `409` cannot be told apart from the caller's own earlier attempt having committed it, so recovery goes through queue redelivery instead, where the step's ownership stamp routes it back to the same invocation. diff --git a/packages/core/src/runtime/constants.test.ts b/packages/core/src/runtime/constants.test.ts index c42038bd45..593db3489a 100644 --- a/packages/core/src/runtime/constants.test.ts +++ b/packages/core/src/runtime/constants.test.ts @@ -398,13 +398,14 @@ describe('getInlineOwnershipLeaseSeconds', () => { describe('pre-claimed inline pairs fit one batch chunk', () => { it('keeps two rows per inline step inside MAX_BATCH_FANOUT_EVENTS', () => { // The suspension fold folds each lazy-inline step into an adjacent - // [step_created, step_started] pair, and the inline slice sorts to the - // front of the batch — so every pair lands in the FIRST chunk exactly - // while two rows per inline step fit inside one chunk. + // [step_created, step_started] pair and commits the pairs in chunk(s) + // of their own, ahead of the plain creates — so every pair lands in ONE + // leading chunk exactly while two rows per inline step fit inside one + // chunk, and the inline bodies gate on a single small commit. // - // Past that, pairs spill into a trailing chunk. `handleSuspension` gates - // its return on every pair-carrying chunk so that degrades safely, but a - // spilled pair costs the caller its claim/body overlap for no reason. + // Past that, pairs spill into a second pair chunk. `handleSuspension` + // gates its return on every pair-carrying chunk so that degrades + // safely, but the bodies then wait for two commits instead of one. // Raising MAX_MAX_INLINE_STEPS therefore has to raise the chunk cap with // it (and re-check the server's per-batch transaction budget). expect(2 * MAX_MAX_INLINE_STEPS).toBeLessThanOrEqual( diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index ef1b37545c..c7c8a4a3cc 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -1736,10 +1736,11 @@ describe('handleSuspension batched fan-out', () => { ownerMessageId: 'msg_owner_1', }); - expect(createBatch).toHaveBeenCalledTimes(1); + // s1/s2 are inline: their pairs form a chunk of their own, adjacent; + // s3's eager create and the wait ride a second chunk — scheduling + // order preserved within each. + expect(createBatch).toHaveBeenCalledTimes(2); const events = createBatch.mock.calls[0][1]; - // s1/s2 are inline: their pairs lead, adjacent; then s3's eager create - // and the wait — scheduling order preserved. expect( events.map((e: { event: { eventType: string } }) => e.event.eventType) ).toEqual([ @@ -1747,14 +1748,18 @@ describe('handleSuspension batched fan-out', () => { 'step_started', 'step_created', 'step_started', - 'step_created', - 'wait_created', ]); expect( events.map( (e: { event: { correlationId: string } }) => e.event.correlationId ) - ).toEqual(['s1', 's1', 's2', 's2', 's3', 'wait_1']); + ).toEqual(['s1', 's1', 's2', 's2']); + expect( + createBatch.mock.calls[1][1].map( + (e: { event: { eventType: string; correlationId: string } }) => + `${e.event.eventType}:${e.event.correlationId}` + ) + ).toEqual(['step_created:s3', 'wait_created:wait_1']); // The created rows carry the input; the started rows are bare claims // stamped with this invocation's ownership and compute instance. const s1Created = events[0].event; @@ -1891,22 +1896,27 @@ describe('handleSuspension batched fan-out', () => { ownerMessageId: 'msg_owner_1', }); + // The pair engages (s2 is its company) but commits in its own chunk; + // s2's eager create rides a sibling chunk. expect( - createBatch.mock.calls[0][1].map( - (e: { event: { eventType: string; correlationId: string } }) => - `${e.event.eventType}:${e.event.correlationId}` + createBatch.mock.calls.map((call) => + call[1].map( + (e: { event: { eventType: string; correlationId: string } }) => + `${e.event.eventType}:${e.event.correlationId}` + ) ) - ).toEqual(['step_created:s1', 'step_started:s1', 'step_created:s2']); + ).toEqual([['step_created:s1', 'step_started:s1'], ['step_created:s2']]); expect(result.inlineClaims.get('s1')?.owned).toBe(true); expect([...result.createdStepCorrelationIds]).toEqual(['s2']); }); it('keeps pairs whole at the chunk boundary (max inline cap)', async () => { // The inline cap clamps at 16, so 16 pairs = exactly 32 rows — one full - // chunk, pairs adjacent throughout — and the eager overflow spills into - // the next call. (Pairs always occupy the head rows, so with cap*2 == - // MAX_BATCH_FANOUT_EVENTS a straddle is structurally unreachable; the - // chunker still refuses to split one should those constants diverge.) + // pair-only chunk, pairs adjacent throughout — and the eager create + // takes its own call. (Pairs never share a chunk with plain creates, + // and with cap*2 == MAX_BATCH_FANOUT_EVENTS a straddle is structurally + // unreachable; the chunker still refuses to split one should those + // constants diverge.) vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '16'); const createBatch = successfulCreateBatch(); const world = createBatchWorld(vi.fn(), createBatch); @@ -1944,6 +1954,127 @@ describe('handleSuspension batched fan-out', () => { expect([...result.createdStepCorrelationIds]).toEqual(['s17']); }); + describe('pairs commit in their own leading chunk', () => { + const shape = (call: unknown[]) => + ( + call[1] as { event: { eventType: string; correlationId: string } }[] + ).map((e) => `${e.event.eventType}:${e.event.correlationId}`); + const ids = (from: number, to: number) => + Array.from({ length: to - from + 1 }, (_, i) => `s${from + i}`); + + it('3 inline + 26 queued: [6 pair rows] then [26 creates]', async () => { + // The pair chunk carries nothing but the pairs, so the inline bodies + // gate on a 6-row commit; the 26 plain creates (which would have + // fit beside them under the 32-row cap) commit in a sibling chunk. + vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '3'); + const createBatch = successfulCreateBatch(); + const world = createBatchWorld(vi.fn(), createBatch); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension( + stepsAndWait(ids(1, 29)), + globalThis + ), + world, + run: slotRun, + ownerMessageId: 'msg_owner_1', + }); + + expect(createBatch).toHaveBeenCalledTimes(2); + expect(shape(createBatch.mock.calls[0])).toEqual([ + 'step_created:s1', + 'step_started:s1', + 'step_created:s2', + 'step_started:s2', + 'step_created:s3', + 'step_started:s3', + ]); + expect(shape(createBatch.mock.calls[1])).toEqual( + ids(4, 29).map((id) => `step_created:${id}`) + ); + expect(result.inlineClaims.size).toBe(3); + for (const claim of result.inlineClaims.values()) { + expect(claim.owned).toBe(true); + } + expect([...result.createdStepCorrelationIds]).toEqual(ids(4, 29)); + }); + + it('3 inline + 40 queued: [6], [32], [8]', async () => { + // Plain creates fill their own chunks to the cap, after the pairs. + vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '3'); + const createBatch = successfulCreateBatch(); + const world = createBatchWorld(vi.fn(), createBatch); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension( + stepsAndWait(ids(1, 43)), + globalThis + ), + world, + run: slotRun, + ownerMessageId: 'msg_owner_1', + }); + + expect(createBatch).toHaveBeenCalledTimes(3); + expect(shape(createBatch.mock.calls[0])).toEqual([ + 'step_created:s1', + 'step_started:s1', + 'step_created:s2', + 'step_started:s2', + 'step_created:s3', + 'step_started:s3', + ]); + expect(shape(createBatch.mock.calls[1])).toEqual( + ids(4, 35).map((id) => `step_created:${id}`) + ); + expect(shape(createBatch.mock.calls[2])).toEqual( + ids(36, 43).map((id) => `step_created:${id}`) + ); + expect(result.inlineClaims.size).toBe(3); + expect([...result.createdStepCorrelationIds]).toEqual(ids(4, 43)); + }); + + it('3 inline + 1 wait, no queued steps: pairs chunk, wait rides its own', async () => { + // No eager steps at all: the pairs still fold (three inline steps + // are each other's company) and the wait, the only plain create, + // commits in a separate one-row chunk rather than padding the pair + // chunk. + vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '3'); + const createBatch = successfulCreateBatch(); + const eventsCreate = vi.fn(); + const world = createBatchWorld(eventsCreate, createBatch); + + const result = await handleSuspension({ + suspension: new WorkflowSuspension( + stepsAndWait(['s1', 's2', 's3'], 'wait_1'), + globalThis + ), + world, + run: slotRun, + ownerMessageId: 'msg_owner_1', + }); + + expect(createBatch).toHaveBeenCalledTimes(2); + expect(shape(createBatch.mock.calls[0])).toEqual([ + 'step_created:s1', + 'step_started:s1', + 'step_created:s2', + 'step_started:s2', + 'step_created:s3', + 'step_started:s3', + ]); + expect(shape(createBatch.mock.calls[1])).toEqual([ + 'wait_created:wait_1', + ]); + expect(eventsCreate).not.toHaveBeenCalled(); + expect(result.inlineClaims.size).toBe(3); + for (const claim of result.inlineClaims.values()) { + expect(claim.owned).toBe(true); + } + expect(result.createdStepCorrelationIds.size).toBe(0); + }); + }); + it('prefers the readback step entity when the World returns one', async () => { vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '1'); const serverStartedAt = new Date('2026-08-14T01:02:03.000Z'); @@ -2080,9 +2211,10 @@ describe('handleSuspension batched fan-out', () => { }); it('returns off the pair chunk; trailing chunks ride deferredBatchWork', async () => { - // 34 steps with a pair: chunk 1 = pair + 30 eager (32 rows), chunk 2 = - // 3 eager. Releasing only chunk 1 must resolve the handler with the - // claims; chunk 2 settles deferredBatchWork later. + // 34 steps with a pair: chunk 1 = the pair alone (2 rows), chunk 2 = + // 32 eager, chunk 3 = 1 eager. Releasing only chunk 1 must resolve + // the handler with the claims; chunks 2 and 3 settle + // deferredBatchWork later. const { createBatch, releases } = gatedCreateBatch(); const { world, queue } = queueWorld(createBatch); const stepIds = Array.from({ length: 34 }, (_, i) => `s${i + 1}`); @@ -2096,28 +2228,37 @@ describe('handleSuspension batched fan-out', () => { allowDeferredBatchWork: true, }); await vi.waitFor(() => { - expect(createBatch).toHaveBeenCalledTimes(2); + expect(createBatch).toHaveBeenCalledTimes(3); }); + expect(createBatch.mock.calls.map((call) => call[1].length)).toEqual([ + 2, 32, 1, + ]); releases[0](); const result = await pending; - // The handler returned with chunk 2 still uncommitted. + // The handler returned with chunks 2 and 3 still uncommitted. expect(result.inlineClaims.get('s1')?.owned).toBe(true); expect(result.deferredBatchWork).toBeDefined(); expect(await probe(result.deferredBatchWork)).toBe('pending'); // Every eager step is claimed for in-flush publishing up front, so // the caller's dispatch pass skips them all. expect(result.queuedStepCorrelationIds.size).toBe(33); + // The pair chunk carries no eager step, so nothing has published yet: + // each plain chunk's messages wait for THAT chunk's commit. + await tick(); + expect(queue).not.toHaveBeenCalled(); - // Chunk 1's publishes fire off its own commit — 30 messages — while - // chunk 2's three wait for theirs. + // Chunk 2's publishes fire off its own commit — 32 messages — while + // chunk 3's one waits for its own. + releases[1](); await vi.waitFor(() => { - expect(queue).toHaveBeenCalledTimes(30); + expect(queue).toHaveBeenCalledTimes(32); }); const publishedNow = queue.mock.calls.map((call) => call[1].stepId); - expect(publishedNow).not.toContain('s33'); + expect(publishedNow).not.toContain('s34'); + expect(await probe(result.deferredBatchWork)).toBe('pending'); - releases[1](); + releases[2](); // biome-ignore lint/style/noNonNullAssertion: asserted defined above await result.deferredBatchWork!; expect(queue).toHaveBeenCalledTimes(33); @@ -2160,6 +2301,8 @@ describe('handleSuspension batched fan-out', () => { const { world } = queueWorld(createBatch); const stepIds = Array.from({ length: 34 }, (_, i) => `s${i + 1}`); + // Chunk 1 = the pair alone (commits at once); chunk 2 = 32 eager + // creates, the one that fails; chunk 3 = the last eager create. const result = await handleSuspension({ suspension: new WorkflowSuspension(stepsAndWait(stepIds), globalThis), world, @@ -2267,8 +2410,8 @@ describe('handleSuspension batched fan-out', () => { // handleSuspension throws, so the pair-chunk failure path has to join // the trailing work itself. let rejectPairChunk: ((err: unknown) => void) | undefined; - let releaseTrailing: (() => void) | undefined; - let trailingSettled = false; + const releaseTrailing: (() => void)[] = []; + let trailingSettled = 0; let call = 0; const createBatch = vi.fn().mockImplementation((_runId, events) => { call += 1; @@ -2278,8 +2421,8 @@ describe('handleSuspension batched fan-out', () => { }); } return new Promise((resolve) => { - releaseTrailing = () => { - trailingSettled = true; + releaseTrailing.push(() => { + trailingSettled += 1; let slot = 100; resolve({ results: events.map(({ event }: { event: object }) => ({ @@ -2287,7 +2430,7 @@ describe('handleSuspension batched fan-out', () => { event: { ...event, eventId: slotToEventId(slot++) }, })), }); - }; + }); }); }); const { world } = queueWorld(createBatch); @@ -2301,23 +2444,26 @@ describe('handleSuspension batched fan-out', () => { stepDispatch: stepDispatch(), allowDeferredBatchWork: true, }); + // Chunk 1 = the pair alone; chunks 2 and 3 = the 33 eager creates. await vi.waitFor(() => { - expect(createBatch).toHaveBeenCalledTimes(2); + expect(createBatch).toHaveBeenCalledTimes(3); }); // biome-ignore lint/style/noNonNullAssertion: set by the first call rejectPairChunk!( new WorkflowWorldError('pair chunk exploded', { status: 500 }) ); - // The rejection must NOT escape while chunk 2 is outstanding. + // The rejection must NOT escape while any trailing chunk is + // outstanding. + expect(await probe(pending)).toBe('pending'); + expect(trailingSettled).toBe(0); + releaseTrailing[0](); expect(await probe(pending)).toBe('pending'); - expect(trailingSettled).toBe(false); - // biome-ignore lint/style/noNonNullAssertion: set by the second call - releaseTrailing!(); + releaseTrailing[1](); await expect(pending).rejects.toMatchObject({ message: expect.stringContaining('pair chunk exploded'), }); - expect(trailingSettled).toBe(true); + expect(trailingSettled).toBe(2); }); it('awaits everything at return without the opt-in', async () => { @@ -2333,12 +2479,15 @@ describe('handleSuspension batched fan-out', () => { stepDispatch: stepDispatch(), }); await vi.waitFor(() => { - expect(createBatch).toHaveBeenCalledTimes(2); + expect(createBatch).toHaveBeenCalledTimes(3); }); releases[0](); - // Chunk 2 unreleased: the handler must still be pending. + // The pair chunk committed but chunks 2 and 3 are unreleased: the + // handler must still be pending. expect(await probe(pending)).toBe('pending'); releases[1](); + expect(await probe(pending)).toBe('pending'); + releases[2](); const result = await pending; expect(result.deferredBatchWork).toBeUndefined(); expect(result.inlineClaims.get('s1')?.owned).toBe(true); diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index c28234802c..cb5870d7dc 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -1134,8 +1134,9 @@ export async function handleSuspension({ // Deterministic position in the batched fold (assigned in stepItems // order, before the concurrent dehydration runs). A pair-folded inline // step occupies two consecutive positions (created row then started - // row), which the flush keeps adjacent and never splits across chunks, - // so a World can fold them into one born-running create. + // row), which the flush keeps adjacent, never splits across chunks, + // and commits in a pair-only chunk ahead of the plain creates, so a + // World can fold them into one born-running create. const pairFolded = inlinePairFoldEligible && lazyInlineCorrelationIds.has(queueItem.correlationId); @@ -1424,7 +1425,9 @@ export async function handleSuspension({ // delivery the way a single-path rejection would. // // Latency shape: only the chunk carrying the pre-claimed inline pairs - // gates the handler's return (the caller starts bodies off its claims). + // gates the handler's return (the caller starts bodies off its claims), + // and that chunk carries NOTHING else, so its commit is as small as the + // inline slice (see the chunking below). // Every other chunk's commit (and every chunk's step-message publishes, // which fire the moment ITS creates are durable) rides // `deferredBatchWork` when the caller opted in, joined before ack. A slow @@ -1485,17 +1488,32 @@ export async function handleSuspension({ const expectedFirstSlot = eventLog ? (maxEventSlot(eventLog.events) ?? 0) + 1 : undefined; - // Pair-aware chunking: a pre-claimed pair's two rows must land in - // the same createBatch call (adjacent, so a World can fold them - // into one born-running create) and never straddle a chunk - // boundary, which would turn the started row into a standalone - // claim racing its own create's commit. - const chunks: (typeof entries)[] = []; - { + // Pair-aware chunking. Two rules: + // + // 1. A pre-claimed pair's two rows must land in the same createBatch + // call (adjacent, so a World can fold them into one born-running + // create) and never straddle a chunk boundary, which would turn + // the started row into a standalone claim racing its own + // create's commit. + // 2. The pairs commit in chunk(s) of THEIR OWN, ahead of the plain + // `step`/`wait` creates, which fill the subsequent chunks. The + // pair chunk is the one commit the inline bodies wait for (see + // `pairCommits` below), and on the Vercel backend its latency + // scales with the transaction's item count: measured on 32-branch + // fan-outs, batches of <=8 events commit in ~56 ms p50 server-side + // where batches of 24-32 events take ~110 ms. Padding the pair + // chunk with plain creates up to the cap therefore held the + // bodies for the plain creates' commit, which nothing else + // needed: a plain create gates only its own queue publish, and + // that fires off whichever sibling chunk carries it. Splitting + // them keeps the pair chunk small (two rows per inline step) and + // lets the plain creates commit concurrently beside it. + const chunkEntries = (list: typeof entries): (typeof entries)[] => { + const out: (typeof entries)[] = []; let current: typeof entries = []; - for (let index = 0; index < entries.length; index++) { - const entry = entries[index]; - const next = entries[index + 1]; + for (let index = 0; index < list.length; index++) { + const entry = list[index]; + const next = list[index + 1]; const pairLead = entry.kind === 'inline-created' && next?.kind === 'inline-started' && @@ -1505,7 +1523,7 @@ export async function handleSuspension({ current.length > 0 && current.length + take > MAX_BATCH_FANOUT_EVENTS ) { - chunks.push(current); + out.push(current); current = []; } current.push(entry); @@ -1514,8 +1532,18 @@ export async function handleSuspension({ index++; } } - if (current.length > 0) chunks.push(current); - } + if (current.length > 0) out.push(current); + return out; + }; + // Both partitions keep `entries`' order, so a pair's two rows + // (consecutive `order` values, enqueued together) stay adjacent in + // the pair partition. + const isPairRow = (entry: (typeof entries)[number]): boolean => + entry.kind === 'inline-created' || entry.kind === 'inline-started'; + const chunks: (typeof entries)[] = [ + ...chunkEntries(entries.filter((entry) => isPairRow(entry))), + ...chunkEntries(entries.filter((entry) => !isPairRow(entry))), + ]; // Steps whose queue messages THIS FLUSH will publish (the eager // creates), recorded before any chunk settles so the caller's // dispatch pass (which runs off the handler's return) skips them. @@ -1773,11 +1801,11 @@ export async function handleSuspension({ // a pair whose commit the caller has not seen yields no // `inlineClaims` entry, so the caller falls back to a lazy // `step_started` that would race this same fold's still-in-flight - // pair for the same step. Today pairs always land in one chunk - // (they sort first, and two rows per inline step fit inside one - // chunk, pinned by constants.test.ts), so this is at most one - // commit; the filter is what keeps the property true if either cap - // moves. + // pair for the same step. The pairs occupy their own leading + // chunk(s), and today that is exactly ONE chunk (two rows per + // inline step fit inside one chunk, pinned by constants.test.ts), + // so this is at most one commit; the filter is what keeps the + // property true if either cap moves. const pairCommits = chunks.flatMap((chunk, index) => chunk.some((entry) => entry.kind === 'inline-started') ? [commits[index]] From 2bd385b1eb5b0c2c26ef92ebb02b9335f85a1721 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 11 Sep 2026 01:34:25 -0700 Subject: [PATCH 2/2] perf(core): fold pairs only with two inline steps; guard a lone plain create With the pairs in a chunk of their own, plain creates alongside them share no round trip with the pairs, so "company" no longer makes a lone pair worth folding. `inlinePairFoldEligible` now requires two or more inline steps: a lone inline step keeps the lazy `step_started` (one row, optimistic-start capable, bump-and-report) while the eager creates beside it still batch. A plain partition of exactly one entry is the batch-of-one case again, so it takes the guarded single create (slot-snapshot params, bump-and-report, same conflict tolerance and `createdStepCorrelationIds` bookkeeping) instead of a one-row createBatch, and its queue publish still waits for that create. Co-Authored-By: Claude Fable 5.1 --- .changeset/batch-pair-chunk.md | 2 +- .../v5/changelog/batched-event-writes.mdx | 2 +- .../src/runtime/suspension-handler.test.ts | 207 +++++++++++++----- .../core/src/runtime/suspension-handler.ts | 97 +++++--- 4 files changed, 221 insertions(+), 87 deletions(-) diff --git a/.changeset/batch-pair-chunk.md b/.changeset/batch-pair-chunk.md index 992c088571..1954f2f6fd 100644 --- a/.changeset/batch-pair-chunk.md +++ b/.changeset/batch-pair-chunk.md @@ -2,4 +2,4 @@ "@workflow/core": patch --- -Commit the batched fan-out's pre-claimed inline `[step_created, step_started]` pairs in their own leading chunk, ahead of the plain step/wait creates, so the write that gates the inline bodies stays small and commits faster. +Commit the batched fan-out's pre-claimed inline `[step_created, step_started]` pairs in their own leading chunk, ahead of the plain step/wait creates, so the write that gates the inline bodies stays small and commits faster. The pairs now fold only with two or more inline steps (a lone inline step keeps the lazy start), and a lone plain create beside them takes the single write. diff --git a/docs/content/docs/v5/changelog/batched-event-writes.mdx b/docs/content/docs/v5/changelog/batched-event-writes.mdx index 5801e6d4b9..edbcfafbb6 100644 --- a/docs/content/docs/v5/changelog/batched-event-writes.mdx +++ b/docs/content/docs/v5/changelog/batched-event-writes.mdx @@ -66,7 +66,7 @@ The contract: **Per-chunk continuation.** Each chunk's follow-on work starts the moment **that chunk** commits, not when the whole fold does: a chunk's step-execution queue messages publish right off its own commit (publish-after-create holds per step), and only the chunk carrying the inline pairs gates the replay's continuation: trailing chunks' commits and publishes are joined before the invocation can acknowledge its message, so the durability contract ("every create durable before ack") is unchanged. -**Pre-claimed inline pairs.** When the fold engages and has company for them (at least two inline steps, or one plus other batchable events), the steps the runtime is about to execute inline join the batch as adjacent `[step_created, step_started]` pairs: the created row carrying the input, the started row a bare ownership-stamped claim the World folds into a born-running create. The pairs commit in a chunk of their own, ahead of the plain `step_created` and `wait_created` chunks, so the write the inline bodies wait for carries only two rows per inline step (a small transaction that commits faster than a full 32-event chunk) while the plain creates commit concurrently beside it. The inline bodies start straight off the pair chunk's commit (in parallel with the queue publishes and the sibling chunks) with no per-step claim POST at all, and a pair that loses its atomic create-claim to a concurrent delivery skips its body exactly as a lost lazy claim does. A lone inline step with nothing else to batch keeps the optimistic lazy-start path, whose claim overlaps the body. +**Pre-claimed inline pairs.** When the fold engages with at least two inline steps, the steps the runtime is about to execute inline join the batch as adjacent `[step_created, step_started]` pairs: the created row carrying the input, the started row a bare ownership-stamped claim the World folds into a born-running create. The pairs commit in a chunk of their own, ahead of the plain `step_created` and `wait_created` chunks, so the write the inline bodies wait for carries only two rows per inline step (a small transaction that commits faster than a full 32-event chunk) while the plain creates commit concurrently beside it. The inline bodies start straight off the pair chunk's commit (in parallel with the queue publishes and the sibling chunks) with no per-step claim POST at all, and a pair that loses its atomic create-claim to a concurrent delivery skips its body exactly as a lost lazy claim does. A lone inline step keeps the optimistic lazy-start path (one row, whose claim overlaps the body) even when eager creates batch beside it: the pairs share no round trip with those creates, so only two or more inline steps make a pair chunk worth the trade. A plain partition of exactly one `step_created` or `wait_created` beside the pairs is written through the ordinary single path rather than a one-row batch, and its queue message still waits for that write. 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. A batch carrying a `step_started` (that is, any batch with inline pairs) is **not** retried in-process on a transport blip: a pair's `409` cannot be told apart from the caller's own earlier attempt having committed it, so recovery goes through queue redelivery instead, where the step's ownership stamp routes it back to the same invocation. diff --git a/packages/core/src/runtime/suspension-handler.test.ts b/packages/core/src/runtime/suspension-handler.test.ts index c7c8a4a3cc..bf9df324c9 100644 --- a/packages/core/src/runtime/suspension-handler.test.ts +++ b/packages/core/src/runtime/suspension-handler.test.ts @@ -1871,9 +1871,10 @@ describe('handleSuspension batched fan-out', () => { ownerMessageId: 'msg_owner_1', }); - // A pair-only batch is the same round trip as the single lazy claim - // but gives up the optimistic claim/body overlap — so nothing is - // written at all here; the deferral stands. + // A pair-only batch of one pair is the same round trip as the single + // lazy claim but gives up the optimistic claim/body overlap and the + // guarded write's bump-and-report — so nothing is written at all + // here; the deferral stands. expect(createBatch).not.toHaveBeenCalled(); expect(eventsCreate).not.toHaveBeenCalled(); expect(result.inlineClaims.size).toBe(0); @@ -1882,13 +1883,14 @@ describe('handleSuspension batched fan-out', () => { ]); }); - it('folds a lone inline pair when an eager sibling already batches', async () => { + it('keeps a lone inline step on the lazy path beside eager creates, which still batch', async () => { + const eventsCreate = vi.fn(); const createBatch = successfulCreateBatch(); - const world = createBatchWorld(vi.fn(), createBatch); + const world = createBatchWorld(eventsCreate, createBatch); const result = await handleSuspension({ suspension: new WorkflowSuspension( - stepsAndWait(['s1', 's2']), + stepsAndWait(['s1', 's2', 's3']), globalThis ), world, @@ -1896,8 +1898,9 @@ describe('handleSuspension batched fan-out', () => { ownerMessageId: 'msg_owner_1', }); - // The pair engages (s2 is its company) but commits in its own chunk; - // s2's eager create rides a sibling chunk. + // Eager company shares no round trip with a pair (the pairs commit in + // a chunk of their own), so it cannot make a lone pair worth folding: + // s1 keeps the lazy `step_started` while s2/s3 batch as before. expect( createBatch.mock.calls.map((call) => call[1].map( @@ -1905,21 +1908,28 @@ describe('handleSuspension batched fan-out', () => { `${e.event.eventType}:${e.event.correlationId}` ) ) - ).toEqual([['step_created:s1', 'step_started:s1'], ['step_created:s2']]); - expect(result.inlineClaims.get('s1')?.owned).toBe(true); - expect([...result.createdStepCorrelationIds]).toEqual(['s2']); + ).toEqual([['step_created:s2', 'step_created:s3']]); + expect(eventsCreate).not.toHaveBeenCalled(); + expect(result.inlineClaims.size).toBe(0); + expect(result.lazyInlineSteps.map((s) => s.correlationId)).toEqual([ + 's1', + ]); + expect([...result.createdStepCorrelationIds]).toEqual(['s2', 's3']); }); it('keeps pairs whole at the chunk boundary (max inline cap)', async () => { // The inline cap clamps at 16, so 16 pairs = exactly 32 rows — one full - // pair-only chunk, pairs adjacent throughout — and the eager create - // takes its own call. (Pairs never share a chunk with plain creates, - // and with cap*2 == MAX_BATCH_FANOUT_EVENTS a straddle is structurally - // unreachable; the chunker still refuses to split one should those - // constants diverge.) + // pair-only chunk, pairs adjacent throughout — and the lone eager + // create takes the guarded single path. (Pairs never share a chunk + // with plain creates, and with cap*2 == MAX_BATCH_FANOUT_EVENTS a + // straddle is structurally unreachable; the chunker still refuses to + // split one should those constants diverge.) vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '16'); const createBatch = successfulCreateBatch(); - const world = createBatchWorld(vi.fn(), createBatch); + const eventsCreate = vi + .fn() + .mockImplementation(async (_runId, event) => ({ event })); + const world = createBatchWorld(eventsCreate, createBatch); const stepIds = Array.from({ length: 17 }, (_, i) => `s${i + 1}`); const result = await handleSuspension({ @@ -1929,7 +1939,7 @@ describe('handleSuspension batched fan-out', () => { ownerMessageId: 'msg_owner_1', }); - expect(createBatch).toHaveBeenCalledTimes(2); + expect(createBatch).toHaveBeenCalledTimes(1); const head = createBatch.mock.calls[0][1]; expect(head).toHaveLength(32); // 16 adjacent created+started pairs, in step order. @@ -1940,13 +1950,15 @@ describe('handleSuspension batched fan-out', () => { head[2 * pair].event.correlationId ); } - const tail = createBatch.mock.calls[1][1]; - expect( - tail.map( - (e: { event: { eventType: string; correlationId: string } }) => - `${e.event.eventType}:${e.event.correlationId}` - ) - ).toEqual(['step_created:s17']); + expect(eventsCreate).toHaveBeenCalledTimes(1); + expect(eventsCreate).toHaveBeenCalledWith( + slotRun.runId, + expect.objectContaining({ + eventType: 'step_created', + correlationId: 's17', + }), + expect.anything() + ); expect(result.inlineClaims.size).toBe(16); for (const claim of result.inlineClaims.values()) { expect(claim.owned).toBe(true); @@ -2034,14 +2046,16 @@ describe('handleSuspension batched fan-out', () => { expect([...result.createdStepCorrelationIds]).toEqual(ids(4, 43)); }); - it('3 inline + 1 wait, no queued steps: pairs chunk, wait rides its own', async () => { - // No eager steps at all: the pairs still fold (three inline steps - // are each other's company) and the wait, the only plain create, - // commits in a separate one-row chunk rather than padding the pair - // chunk. + it('3 inline + 1 wait, no queued steps: pairs chunk, wait takes the single path', async () => { + // No eager steps at all: the pairs still fold (three inline steps) + // and the wait, the only plain create, is a plain partition of one, + // so it takes the guarded single write rather than padding the pair + // chunk or going out as a one-row batch. vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '3'); const createBatch = successfulCreateBatch(); - const eventsCreate = vi.fn(); + const eventsCreate = vi + .fn() + .mockImplementation(async (_runId, event) => ({ event })); const world = createBatchWorld(eventsCreate, createBatch); const result = await handleSuspension({ @@ -2054,7 +2068,7 @@ describe('handleSuspension batched fan-out', () => { ownerMessageId: 'msg_owner_1', }); - expect(createBatch).toHaveBeenCalledTimes(2); + expect(createBatch).toHaveBeenCalledTimes(1); expect(shape(createBatch.mock.calls[0])).toEqual([ 'step_created:s1', 'step_started:s1', @@ -2063,10 +2077,15 @@ describe('handleSuspension batched fan-out', () => { 'step_created:s3', 'step_started:s3', ]); - expect(shape(createBatch.mock.calls[1])).toEqual([ - 'wait_created:wait_1', - ]); - expect(eventsCreate).not.toHaveBeenCalled(); + expect(eventsCreate).toHaveBeenCalledTimes(1); + expect(eventsCreate).toHaveBeenCalledWith( + slotRun.runId, + expect.objectContaining({ + eventType: 'wait_created', + correlationId: 'wait_1', + }), + expect.anything() + ); expect(result.inlineClaims.size).toBe(3); for (const claim of result.inlineClaims.values()) { expect(claim.owned).toBe(true); @@ -2076,7 +2095,7 @@ describe('handleSuspension batched fan-out', () => { }); it('prefers the readback step entity when the World returns one', async () => { - vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '1'); + vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '2'); const serverStartedAt = new Date('2026-08-14T01:02:03.000Z'); let slot = 30; const createBatch = vi @@ -2211,13 +2230,14 @@ describe('handleSuspension batched fan-out', () => { }); it('returns off the pair chunk; trailing chunks ride deferredBatchWork', async () => { - // 34 steps with a pair: chunk 1 = the pair alone (2 rows), chunk 2 = - // 32 eager, chunk 3 = 1 eager. Releasing only chunk 1 must resolve - // the handler with the claims; chunks 2 and 3 settle + // 35 steps with two pairs: chunk 1 = the pairs alone (4 rows), chunk + // 2 = 32 eager, chunk 3 = 1 eager. Releasing only chunk 1 must + // resolve the handler with the claims; chunks 2 and 3 settle // deferredBatchWork later. + vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '2'); const { createBatch, releases } = gatedCreateBatch(); const { world, queue } = queueWorld(createBatch); - const stepIds = Array.from({ length: 34 }, (_, i) => `s${i + 1}`); + const stepIds = Array.from({ length: 35 }, (_, i) => `s${i + 1}`); const pending = handleSuspension({ suspension: new WorkflowSuspension(stepsAndWait(stepIds), globalThis), @@ -2231,13 +2251,14 @@ describe('handleSuspension batched fan-out', () => { expect(createBatch).toHaveBeenCalledTimes(3); }); expect(createBatch.mock.calls.map((call) => call[1].length)).toEqual([ - 2, 32, 1, + 4, 32, 1, ]); releases[0](); const result = await pending; // The handler returned with chunks 2 and 3 still uncommitted. expect(result.inlineClaims.get('s1')?.owned).toBe(true); + expect(result.inlineClaims.get('s2')?.owned).toBe(true); expect(result.deferredBatchWork).toBeDefined(); expect(await probe(result.deferredBatchWork)).toBe('pending'); // Every eager step is claimed for in-flush publishing up front, so @@ -2255,7 +2276,7 @@ describe('handleSuspension batched fan-out', () => { expect(queue).toHaveBeenCalledTimes(32); }); const publishedNow = queue.mock.calls.map((call) => call[1].stepId); - expect(publishedNow).not.toContain('s34'); + expect(publishedNow).not.toContain('s35'); expect(await probe(result.deferredBatchWork)).toBe('pending'); releases[2](); @@ -2275,6 +2296,85 @@ describe('handleSuspension batched fan-out', () => { ); }); + it('2 inline + 1 eager: pair chunk via createBatch, the eager create guarded, its publish after it', async () => { + // The plain partition is exactly one entry, so it takes the guarded + // single write (`events.create` with the slot-snapshot params) rather + // than a one-row createBatch, and its queue message still waits for + // THAT create: publish-after-create holds for the single too. + vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '2'); + const { createBatch, releases } = gatedCreateBatch(); + let releaseSingle: (() => void) | undefined; + const eventsCreate = vi.fn().mockImplementation( + (_runId, event) => + new Promise((resolve) => { + releaseSingle = () => resolve({ event }); + }) + ); + const queue = vi.fn().mockResolvedValue({ messageId: 'msg_q' }); + const world = { + events: { create: eventsCreate, createBatch }, + queue, + getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), + } as unknown as World; + + const pending = handleSuspension({ + suspension: new WorkflowSuspension( + stepsAndWait(['s1', 's2', 's3']), + globalThis + ), + world, + run: slotRun, + ownerMessageId: 'msg_owner_1', + stepDispatch: stepDispatch(), + allowDeferredBatchWork: true, + }); + // Both writes are in flight together: one createBatch carrying the + // two pairs and nothing else, one guarded single for s3. + await vi.waitFor(() => { + expect(createBatch).toHaveBeenCalledTimes(1); + expect(eventsCreate).toHaveBeenCalledTimes(1); + }); + expect( + createBatch.mock.calls[0][1].map( + (e: { event: { eventType: string; correlationId: string } }) => + `${e.event.eventType}:${e.event.correlationId}` + ) + ).toEqual([ + 'step_created:s1', + 'step_started:s1', + 'step_created:s2', + 'step_started:s2', + ]); + expect(eventsCreate).toHaveBeenCalledWith( + slotRun.runId, + expect.objectContaining({ + eventType: 'step_created', + correlationId: 's3', + }), + expect.anything() + ); + + // The pair chunk alone releases the handler; s3's create is trailing. + releases[0](); + const result = await pending; + expect(result.inlineClaims.get('s1')?.owned).toBe(true); + expect(result.inlineClaims.get('s2')?.owned).toBe(true); + expect([...result.queuedStepCorrelationIds]).toEqual(['s3']); + expect(result.createdStepCorrelationIds.size).toBe(0); + expect(await probe(result.deferredBatchWork)).toBe('pending'); + await tick(); + expect(queue).not.toHaveBeenCalled(); + + // s3's message goes out only once its own create is durable. + // biome-ignore lint/style/noNonNullAssertion: set by the single create + releaseSingle!(); + // biome-ignore lint/style/noNonNullAssertion: asserted defined above + await result.deferredBatchWork!; + expect(queue).toHaveBeenCalledTimes(1); + expect(queue.mock.calls[0][1].stepId).toBe('s3'); + expect([...result.createdStepCorrelationIds]).toEqual(['s3']); + }); + it('surfaces a trailing-chunk failure through deferredBatchWork, not the return', async () => { let call = 0; let releaseFailure: (() => void) | undefined; @@ -2298,11 +2398,12 @@ describe('handleSuspension batched fan-out', () => { })), }); }); + vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '2'); const { world } = queueWorld(createBatch); - const stepIds = Array.from({ length: 34 }, (_, i) => `s${i + 1}`); + const stepIds = Array.from({ length: 35 }, (_, i) => `s${i + 1}`); - // Chunk 1 = the pair alone (commits at once); chunk 2 = 32 eager - // creates, the one that fails; chunk 3 = the last eager create. + // Chunk 1 = the two pairs alone (commits at once); chunk 2 = 32 + // eager creates, the one that fails; chunk 3 = the last eager create. const result = await handleSuspension({ suspension: new WorkflowSuspension(stepsAndWait(stepIds), globalThis), world, @@ -2360,8 +2461,11 @@ describe('handleSuspension batched fan-out', () => { getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined), } as unknown as World; + // s1/s2 pair-fold (cap 2); s5 is finalized sequentially; the 32 + // healthy eager creates fill the second call, the one that fails. + vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '2'); const pending = stepsAndWait( - Array.from({ length: 34 }, (_, i) => `s${i + 1}`) + Array.from({ length: 35 }, (_, i) => `s${i + 1}`) ) as Map; // biome-ignore lint/style/noNonNullAssertion: seeded above pending.get('s5')!.args = [new Unserializable()]; @@ -2433,8 +2537,9 @@ describe('handleSuspension batched fan-out', () => { }); }); }); + vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '2'); const { world } = queueWorld(createBatch); - const stepIds = Array.from({ length: 34 }, (_, i) => `s${i + 1}`); + const stepIds = Array.from({ length: 35 }, (_, i) => `s${i + 1}`); const pending = handleSuspension({ suspension: new WorkflowSuspension(stepsAndWait(stepIds), globalThis), @@ -2444,7 +2549,8 @@ describe('handleSuspension batched fan-out', () => { stepDispatch: stepDispatch(), allowDeferredBatchWork: true, }); - // Chunk 1 = the pair alone; chunks 2 and 3 = the 33 eager creates. + // Chunk 1 = the two pairs alone; chunks 2 and 3 = the 33 eager + // creates. await vi.waitFor(() => { expect(createBatch).toHaveBeenCalledTimes(3); }); @@ -2467,9 +2573,10 @@ describe('handleSuspension batched fan-out', () => { }); it('awaits everything at return without the opt-in', async () => { + vi.stubEnv('WORKFLOW_MAX_INLINE_STEPS', '2'); const { createBatch, releases } = gatedCreateBatch(); const { world } = queueWorld(createBatch); - const stepIds = Array.from({ length: 34 }, (_, i) => `s${i + 1}`); + const stepIds = Array.from({ length: 35 }, (_, i) => `s${i + 1}`); const pending = handleSuspension({ suspension: new WorkflowSuspension(stepsAndWait(stepIds), globalThis), diff --git a/packages/core/src/runtime/suspension-handler.ts b/packages/core/src/runtime/suspension-handler.ts index cb5870d7dc..3ff3014eb1 100644 --- a/packages/core/src/runtime/suspension-handler.ts +++ b/packages/core/src/runtime/suspension-handler.ts @@ -990,7 +990,7 @@ export async function handleSuspension({ failedStepCorrelationIds.add(queueItem.correlationId); // Release the inline slot bookkeeping: the step never runs, so it must // not appear in the rebuilt `lazyInlineSteps`. (Its slot in the first-N - // selection and in `inlinePairFoldEligible`'s arithmetic was consumed + // selection and in `inlinePairFoldEligible`'s count was consumed // before dehydration could reveal the failure, inherent to selecting // before serializing, and bounded to one wasted slot on a pass that // ends in a forced replay anyway.) @@ -1088,28 +1088,27 @@ export async function handleSuspension({ // Pre-claimed inline pairs: fold each lazy-inline step's deferred // `step_created` (carrying its input) AND its `step_started` claim (bare, - // ownership-stamped) into the batch, so the whole fan-out (the inline - // steps' claims included) commits in the one durable write and the caller + // ownership-stamped) into the batch, so the inline steps' claims commit in + // one durable write (the pair chunk, see the flush below) and the caller // starts the bodies straight off that commit instead of posting one claim - // per inline step. The lone-inline case (nothing else to batch with) is - // excluded: a pair-only batch costs the same round trip as the single lazy - // claim while giving up the optimistic claim/body overlap and the - // bump-and-report that `createGuarded` provides, so it stays on the lazy - // path. Requires the caller's `ownerMessageId`: the started row must - // stamp ownership exactly like the lazy claim it replaces (and a caller - // that does not inline-execute never provides one). - const uncreatedWaitCount = waitItems.filter( - (item) => !item.hasCreatedEvent - ).length; + // per inline step. Requires TWO or more inline steps, and nothing else: + // the pairs commit in a chunk of their own, so plain creates alongside + // them share no round trip with the pairs and cannot make a lone pair + // worth folding. A lone inline step is better served by the lazy + // `step_started`: one row instead of two, optimistic-start capable (the + // claim overlaps the body), with the slot-snapshot params and + // bump-and-report `createGuarded` provides, all of which a pair-only + // batch of one pair gives up for the same round trip. Two or more inline + // steps become ONE pair chunk instead of N parallel lazy claims, which is + // the saving. A lone inline step alongside eager creates therefore takes + // the lazy path while the creates still batch. Also requires the caller's + // `ownerMessageId`: the started row must stamp ownership exactly like the + // lazy claim it replaces (and a caller that does not inline-execute never + // provides one). const inlinePairFoldEligible = batchFanoutEligible && ownerMessageId !== undefined && - lazyInlineCorrelationIds.size > 0 && - (lazyInlineCorrelationIds.size >= 2 || - stepsNeedingCreation.size - - lazyInlineCorrelationIds.size + - uncreatedWaitCount >= - 1); + lazyInlineCorrelationIds.size >= 2; const inlineClaims: SuspensionHandlerResult['inlineClaims'] = new Map(); // The trace carrier for resilient step dispatches, resolved at most once per @@ -1448,13 +1447,20 @@ 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; + // The ordinary single path for a lone plain entry: 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 is written the way it would have + // been without the fold, with the same conflict tolerance and + // ownership bookkeeping. Used when the whole fold is one entry + // (below) and when the plain partition beside the pairs is one + // entry (the chunking further down). Beside a pair chunk it commits + // concurrently with the handler's return, like a trailing chunk; the + // inline bodies do not depend on it (executor writes carry no slot + // snapshot, so there is no position for it to move). + const commitSingle = async ( + entry: (typeof entries)[number] + ): Promise => { try { await createGuarded(entry.event, { requestId }); if (entry.kind === 'step') { @@ -1476,6 +1482,9 @@ export async function handleSuspension({ throw err; } } + }; + if (entries.length === 1) { + await commitSingle(entries[0]); return; } // Seed for the foreign-interleaving diagnostic below. With chunks @@ -1540,10 +1549,24 @@ export async function handleSuspension({ // the pair partition. const isPairRow = (entry: (typeof entries)[number]): boolean => entry.kind === 'inline-created' || entry.kind === 'inline-started'; - const chunks: (typeof entries)[] = [ - ...chunkEntries(entries.filter((entry) => isPairRow(entry))), - ...chunkEntries(entries.filter((entry) => !isPairRow(entry))), - ]; + const pairChunks = chunkEntries( + entries.filter((entry) => isPairRow(entry)) + ); + const plainEntries = entries.filter((entry) => !isPairRow(entry)); + // A plain partition of exactly ONE entry is the batch-of-one case + // again, now that the pairs no longer share its chunk: it takes the + // single path (`commitSingle`) rather than a one-row createBatch, + // and is carried as a one-entry chunk so the per-chunk publish and + // settle machinery below treats it like any sibling. (A one-row + // remainder of a LARGER plain partition still batches: it is the + // pre-split behavior, unchanged here.) + const plainChunks = + plainEntries.length === 1 + ? [plainEntries] + : chunkEntries(plainEntries); + const singleChunk = + plainEntries.length === 1 ? plainChunks[0] : undefined; + const chunks: (typeof entries)[] = [...pairChunks, ...plainChunks]; // Steps whose queue messages THIS FLUSH will publish (the eager // creates), recorded before any chunk settles so the caller's // dispatch pass (which runs off the handler's return) skips them. @@ -1751,11 +1774,15 @@ export async function handleSuspension({ ); }; - // Launch every chunk's POST now; chain each chunk's publishes on its - // OWN commit. A chunk whose commit rejected keeps its messages - // unsent (the rejection fails the delivery; redelivery re-creates - // and re-dispatches, deduped by the idempotency keys). - const commits = chunks.map((chunk) => commitChunk(chunk)); + // Launch every chunk's POST now (the lone plain entry's guarded + // single create included); chain each chunk's publishes on its OWN + // commit, so publish-after-create holds for the single too. A chunk + // whose commit rejected keeps its messages unsent (the rejection + // fails the delivery; redelivery re-creates and re-dispatches, + // deduped by the idempotency keys). + const commits = chunks.map((chunk) => + chunk === singleChunk ? commitSingle(chunk[0]) : commitChunk(chunk) + ); const publishes = chunks.map(async (chunk, index) => { await commits[index]; await publishChunkSteps(chunk);