perf(core): don't re-publish step messages this invocation already published

.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).

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

packages/core/src/runtime.ts

@@ -2773,6 +2773,31 @@ export function workflowEntrypoint(
                   // continuing in-process for each.
                   const continuedHookIds = new Set<string>();
 
+                  // 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<string>();
+
                   // 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',

packages/core/src/telemetry/semantic-conventions.ts

@@ -237,6 +237,16 @@ export const WorkflowBackstopWakesArmed = SemanticConvention<number>(
   '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<number>(
+  'workflow.dispatch.republish_skipped'
+);
+
 // Route attributes
 
 /** The workflow runtime route being handled */