From a1d0ea6ec804ff3d2542d389392d49638e368d58 Mon Sep 17 00:00:00 2001 From: Pranay Prakash Date: Fri, 11 Sep 2026 01:22:56 -0700 Subject: [PATCH] perf(core): don't re-publish step messages this invocation already published On a fan-out that runs some steps inline, the orchestrator invocation publishes the queued siblings' step-execution messages, runs its inline steps, falls back into the replay loop, reloads the log, and its pending-step dispatch pass publishes every one of those messages again (measured 19-28 re-sends per pass on a 32-branch fan-out, ~400 ms after the originals). The queue dedupes them by idempotency key, but the sends still cost round-trips on the shared connection pool and hold the invocation open past its useful work. Track, per delivery, the correlation ids this invocation has already published a step message for (the suspension handler's resilient publishes plus the dispatch pass's own immediate enqueues) and skip the immediate re-enqueue for those on later passes, unless a step_retrying has been observed since (a new schedule). The set is invocation-scoped and never derived from the log: a step_created does not prove the message was ever sent, so a different delivery still re-enqueues unconditionally. Reported as a debug log and the `workflow.dispatch.republish_skipped` span attribute. The QuickJS engine already keeps the equivalent invocation-scoped `queuedStepIds` set, so it is unchanged. Co-Authored-By: Claude Fable 5.1 --- .changeset/dispatch-skip-republished.md | 5 + packages/core/src/runtime.test.ts | 125 ++++++++++++++++-- packages/core/src/runtime.ts | 83 +++++++++++- .../src/telemetry/semantic-conventions.ts | 10 ++ 4 files changed, 206 insertions(+), 17 deletions(-) create mode 100644 .changeset/dispatch-skip-republished.md diff --git a/.changeset/dispatch-skip-republished.md b/.changeset/dispatch-skip-republished.md new file mode 100644 index 0000000000..80750536ab --- /dev/null +++ b/.changeset/dispatch-skip-republished.md @@ -0,0 +1,5 @@ +--- +"@workflow/core": patch +--- + +Skip re-publishing a pending step's execution message on a later replay pass when this same invocation already published it (a fresh delivery still re-enqueues unconditionally). diff --git a/packages/core/src/runtime.test.ts b/packages/core/src/runtime.test.ts index 763a086bc1..d533e4af16 100644 --- a/packages/core/src/runtime.test.ts +++ b/packages/core/src/runtime.test.ts @@ -1468,8 +1468,20 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { queueName: string, message: any ) => Promise<{ messageId: null }>; + /** + * Record a `step_retrying` for the QUEUED step right after its eager + * `step_created`, so the post-inline replay pass observes the step as + * `sawRetrying` (a retry handoff happened since this invocation + * published it). + */ + retryingQueuedStep?: boolean; + /** `specVersion` to stamp on the run (gates resilient step dispatch). */ + runSpecVersion?: number; }) { const workflowRun = await makeRunningRun(opts.runId); + if (opts.runSpecVersion !== undefined) { + (workflowRun as any).specVersion = opts.runSpecVersion; + } const order: string[] = []; // Start from a clean slate so the rejection check only observes promises @@ -1512,7 +1524,16 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { // It must be durably created before its dispatch send — the ordering // assertion below checks step_created precedes queue_dispatch_start. order.push('step_created'); - return { event: recordEvent(data) }; + const created = recordEvent(data); + if (opts.retryingQueuedStep) { + recordEvent({ + eventType: 'step_retrying', + specVersion: SPEC_VERSION_CURRENT, + correlationId: data.correlationId, + eventData: { stepName: data.eventData?.stepName }, + }); + } + return { event: created }; } if (data.eventType === 'step_started') { // The inline step's lazy step_started creates the step on the fly: @@ -1549,9 +1570,12 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { } ); + // Every step-execution send (message carrying a stepId), in order. + const stepIdSends: string[] = []; const queue = vi.fn(async (queueName: string, message: any) => { // Only the step-dispatch send carries a stepId; ignore other sends. if (message && typeof message === 'object' && 'stepId' in message) { + stepIdSends.push(message.stepId); order.push('queue_dispatch_start'); const result = await opts.queueImpl(queueName, message); order.push('queue_dispatch_done'); @@ -1560,6 +1584,16 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { return { messageId: null }; }); + // Return the accumulated event log so replay converges: a later loop + // iteration sees the inline step completed and the queued step already + // created (so neither is re-run), and the handler returns instead of + // re-suspending forever. + const eventsList = vi.fn(async () => ({ + data: [...durableEvents], + hasMore: false, + cursor: 'cursor_test', + })); + setWorld({ specVersion: SPEC_VERSION_CURRENT, getDeploymentId: vi.fn(async () => workflowRun.deploymentId), @@ -1587,15 +1621,7 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { ), events: { create: eventsCreate, - // Return the accumulated event log so replay converges: a later loop - // iteration sees the inline step completed and the queued step already - // created (so neither is re-run), and the handler returns instead of - // re-suspending forever. - list: vi.fn(async () => ({ - data: [...durableEvents], - hasMore: false, - cursor: 'cursor_test', - })), + list: eventsList, }, runs: { get: vi.fn(async () => workflowRun), @@ -1618,6 +1644,8 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { handlerPromise, order, queue, + eventsList, + stepIdSends, createdEventParams, stepStartedParams, }; @@ -1715,6 +1743,83 @@ describe('workflowEntrypoint step-dispatch ack ordering', () => { expect(await anyWaitUntilPromiseRejected()).toBe(false); }); + it('publishes each queued step message once per invocation: the post-inline replay pass re-publishes nothing', async () => { + // Pass 1 queues `addB` (one step-execution send) and runs `add` inline. + // Once `add` completes the loop reloads the log and replays again; the + // second pass finds `addB` still pending. That pass must NOT send its + // message a second time: this invocation already did, and the queue + // would only dedupe the repeat after a wasted round-trip. + const { handlerPromise, order, eventsList, stepIdSends } = + await driveHandler({ + runId: 'wrun_no_republish', + queueImpl: async () => ({ messageId: null }), + }); + + const res = (await handlerPromise) as Response; + expect(res.status).toBe(204); + + // The second replay pass really happened (a fresh events.list after the + // inline step completed), and it observed the queued step as pending + // (eagerly created, never completed). + expect(eventsList.mock.calls.length).toBeGreaterThanOrEqual(2); + expect(order).toContain('step_created'); + // Exactly one step-execution publish for the whole invocation. + expect(stepIdSends).toHaveLength(1); + }); + + it('still re-enqueues a step it published once a step_retrying has been observed', async () => { + // A `step_retrying` recorded after this invocation's publish means the + // step is now on a retry schedule the earlier message does not cover, so + // the invocation-local "already published" knowledge must not suppress + // the later pass's enqueue (the idempotency key still dedupes if the + // handoff message is in flight). + const { handlerPromise, stepIdSends } = await driveHandler({ + runId: 'wrun_republish_after_retrying', + queueImpl: async () => ({ messageId: null }), + retryingQueuedStep: true, + }); + + const res = (await handlerPromise) as Response; + expect(res.status).toBe(204); + // Pass 1 publish + the pass-2 re-enqueue for the retrying step, same + // correlation id both times. + expect(stepIdSends).toHaveLength(2); + expect(new Set(stepIdSends).size).toBe(1); + }); + + it("counts the suspension handler's resilient publishes as already published", async () => { + // With resilient step dispatch the suspension handler publishes the + // queued step itself (create + queue in parallel, message carrying + // stepInput) and reports it in queuedStepCorrelationIds. That publish + // must seed the invocation's published set too, so the post-inline + // replay pass does not send it again. + process.env.WORKFLOW_RESILIENT_STEP_DISPATCH = '1'; + try { + const { handlerPromise, eventsList, stepIdSends, queue } = + await driveHandler({ + runId: 'wrun_no_republish_resilient', + queueImpl: async () => ({ messageId: null }), + runSpecVersion: SPEC_VERSION_CURRENT, + }); + + const res = (await handlerPromise) as Response; + expect(res.status).toBe(204); + + expect(eventsList.mock.calls.length).toBeGreaterThanOrEqual(2); + // The one send came from the suspension handler (it carries the + // resilient stepInput), and nothing re-published it afterwards. + const stepSends = queue.mock.calls.filter( + ([, message]: any[]) => + message && typeof message === 'object' && 'stepId' in message + ); + expect(stepSends).toHaveLength(1); + expect(stepSends[0][1]).toHaveProperty('stepInput'); + expect(stepIdSends).toHaveLength(1); + } finally { + delete process.env.WORKFLOW_RESILIENT_STEP_DISPATCH; + } + }); + it('runs BOTH parallel steps inline (none queued) when the inline cap allows it', async () => { // Override the per-suite cap of 1: with a cap of 3 both `add` and `addB` // are deferred and run inline via lazy step_started, so neither is eagerly diff --git a/packages/core/src/runtime.ts b/packages/core/src/runtime.ts index 51f013bd8a..a134617cdc 100644 --- a/packages/core/src/runtime.ts +++ b/packages/core/src/runtime.ts @@ -2773,6 +2773,31 @@ export function workflowEntrypoint( // continuing in-process for each. const continuedHookIds = new Set(); + // Steps THIS delivery has already published a + // step-execution message for, across its replay passes: + // the suspension handler's resilient publishes + // (`queuedStepCorrelationIds`) plus the dispatch pass's + // own immediate enqueues. A fan-out that runs some steps + // inline falls back into this loop once they finish, + // reloads the log, and finds the queued siblings still + // pending; without this set the next dispatch pass sends + // every one of their messages again. The queue dedupes + // them by idempotency key, but each redundant send still + // costs a round-trip on the shared connection pool and + // holds the invocation open past its useful work + // (measured: 19-28 re-sends per pass on a 32-branch + // fan-out, ~400 ms after the originals). + // + // Deliberately invocation-scoped, never derived from the + // log: a `step_created` in the log proves the step was + // created, not that its message was ever sent (the + // publisher may have crashed between the two). So a + // DIFFERENT invocation (crash-recovery redelivery, a + // concurrent wake) re-enqueues unconditionally, exactly + // as before, and only knowledge of this process's own + // sends is used to skip. Dies with this delivery. + const publishedStepCorrelationIds = new Set(); + // Main replay loop while (true) { loopIteration++; @@ -3848,6 +3873,13 @@ export function workflowEntrypoint( const ownedRecoverySteps: StepInvocationQueueItem[] = []; let backstopWakesArmed = 0; + // Immediate re-enqueues suppressed because this + // invocation already published the step's message + // on an earlier pass. See publishedStepCorrelationIds. + let republishesSkipped = 0; + for (const correlationId of suspensionResult.queuedStepCorrelationIds) { + publishedStepCorrelationIds.add(correlationId); + } // TTR hand-off. The measurement may only go to an // execution that will actually ATTEMPT the next // durable step, and the loop below is what decides @@ -3894,12 +3926,14 @@ export function workflowEntrypoint( continue; } // Already published by the suspension handler's - // resilient dispatch (create + queue in parallel, - // message carrying `stepInput`). A re-publish here - // would dedupe on the idempotency key anyway, but - // skip the wasted round-trip. Ownership never - // applies to these: they were created this pass, so - // no step_started stamp can exist yet. + // resilient dispatch THIS pass (create + queue in + // parallel, message carrying `stepInput`). A + // re-publish here would dedupe on the idempotency + // key anyway, but skip the wasted round-trip. + // Ownership never applies to these: they were + // created this pass, so no step_started stamp can + // exist yet. (Earlier passes' publishes are + // covered by publishedStepCorrelationIds below.) if ( suspensionResult.queuedStepCorrelationIds.has( step.correlationId @@ -3956,6 +3990,25 @@ export function workflowEntrypoint( ); continue; } + // Already published by THIS invocation on an + // earlier pass (see publishedStepCorrelationIds): + // the message is in the queue and the send is + // joined below, so a repeat buys nothing. A + // `step_retrying` observed since is a new + // schedule (the retry handoff), so it is not + // covered by the earlier publish and re-enqueues + // as before. Checked before the TTR hand-off so a + // skipped step never consumes the measurement. + if ( + publishedStepCorrelationIds.has( + step.correlationId + ) && + step.sawRetrying !== true + ) { + republishesSkipped++; + continue; + } + publishedStepCorrelationIds.add(step.correlationId); // This step IS being attempted, by the invocation // that picks the message up. Consume the tracking // here, for the first such step and no other. @@ -4098,7 +4151,8 @@ export function workflowEntrypoint( // delivery of this message died mid-step-body. if ( backstopWakesArmed > 0 || - ownedRecoverySteps.length > 0 + ownedRecoverySteps.length > 0 || + republishesSkipped > 0 ) { span?.setAttributes({ ...(ownedRecoverySteps.length > 0 @@ -4111,8 +4165,23 @@ export function workflowEntrypoint( backstopWakesArmed ) : {}), + ...(republishesSkipped > 0 + ? Attribute.WorkflowDispatchRepublishSkipped( + republishesSkipped + ) + : {}), }); } + if (republishesSkipped > 0) { + runtimeLogger.debug( + 'Skipped re-publishing step messages this invocation already published on an earlier replay pass', + { + workflowRunId: runId, + loopIteration, + republishesSkipped, + } + ); + } if (ownedRecoverySteps.length > 0) { runtimeLogger.warn( 'Re-executing inline steps owned by this queue message — a previous delivery crashed mid-body and this redelivery is recovering them', diff --git a/packages/core/src/telemetry/semantic-conventions.ts b/packages/core/src/telemetry/semantic-conventions.ts index dbd0f7b5ae..f4a2056cfc 100644 --- a/packages/core/src/telemetry/semantic-conventions.ts +++ b/packages/core/src/telemetry/semantic-conventions.ts @@ -237,6 +237,16 @@ export const WorkflowBackstopWakesArmed = SemanticConvention( 'workflow.inline_ownership.backstop_wakes_armed' ); +/** + * Number of pending steps whose immediate step-execution enqueue this replay + * pass skipped because THIS invocation already published that step's message + * on an earlier pass (a fan-out that ran inline steps and replayed again). + * Invocation-local knowledge only: a fresh delivery never skips. + */ +export const WorkflowDispatchRepublishSkipped = SemanticConvention( + 'workflow.dispatch.republish_skipped' +); + // Route attributes /** The workflow runtime route being handled */