fix(world-local,core): stop re-executing in-flight inline steps on redelivery

.changeset/local-queue-no-deadline-single-flight.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Stop re-executing a running lazy or pre-claimed inline step when its message is redelivered.

packages/core/src/runtime.ts

@@ -5093,21 +5093,27 @@ export function workflowEntrypoint(
                             // these bodies until they settle. See
                             // assertNoInFlightOwnedSteps.
                             inFlightOwnedSteps.add(s.correlationId);
-                            // Lazy and pre-claimed steps are brand-new
-                            // (their create-claim is the exactly-once gate),
-                            // but an owned-recovery step already exists and
-                            // its delayed backstop message may fire mid-body
-                            // in this same process, so route those through
-                            // the in-process single-flight.
-                            const executed =
-                              s.lazyStepInput === undefined &&
-                              s.preclaimedStart === undefined
-                                ? runStepSingleFlight(
-                                    runId,
-                                    s.correlationId,
-                                    run
-                                  )
-                                : run();
+                            // Every inline body runs through the in-process
+                            // single-flight. An owned-recovery step's delayed
+                            // backstop may fire mid-body in this process, and
+                            // a lazy or pre-claimed step's create-claim is not
+                            // an exactly-once gate for side effects either:
+                            // the optimistic start runs the body before the
+                            // claim settles, so a redelivery of this same
+                            // message (a transport timeout re-entering turbo
+                            // at attempt 1, workflow#3909) would run it again
+                            // concurrently. Fresh claims overlap during
+                            // ordinary wake replays, so they log at debug,
+                            // matching quickjs-entrypoint.
+                            const isFreshClaim =
+                              s.lazyStepInput !== undefined ||
+                              s.preclaimedStart !== undefined;
+                            const executed = runStepSingleFlight(
+                              runId,
+                              s.correlationId,
+                              run,
+                              isFreshClaim ? 'debug' : 'warn'
+                            );
                             return executed.finally(() =>
                               inFlightOwnedSteps.delete(s.correlationId)
                             );

workbench/vitest/test/inline-step-ownership.test.ts

@@ -4,7 +4,11 @@ import { join } from 'node:path';
 import { waitForHook } from '@workflow/vitest';
 import { describe, expect, it } from 'vitest';
 import { resumeHook, start } from 'workflow/api';
-import { inlineStepDuringHookResumeWorkflow } from '../workflows/inline-step-ownership.js';
+import { getWorld } from 'workflow/runtime';
+import {
+  inlineStepDuringHookResumeWorkflow,
+  inlineStepDuringRedeliveryWorkflow,
+} from '../workflows/inline-step-ownership.js';
 
 describe('inline step ownership (#2780)', () => {
   // Issue #2780: an inline step has no queue message of its own, so a wake
@@ -42,3 +46,51 @@ describe('inline step ownership (#2780)', () => {
     ).toHaveLength(1);
   }, 60_000);
 });
+
+describe('inline step redelivery (#3909)', () => {
+  // Issue #3909: world-local's queue used to give up on a delivery after 30s
+  // and redeliver the same message while its handler was still running a
+  // lazy inline step. The redelivery re-entered turbo (attempt 1), skipped the
+  // event log, and ran the step body again concurrently. Simulate that
+  // redelivery by re-sending the run's start message mid-body: the in-process
+  // single-flight must absorb it so the body is entered exactly once.
+  it('a redelivered start message mid-inline-step does not re-execute the step body', async () => {
+    const dir = await mkdtemp(join(tmpdir(), 'wf-3909-'));
+    const markerPath = join(dir, 'marker.log');
+
+    const world = await getWorld();
+    const originalQueue = world.queue;
+    const redelivered: Promise<unknown>[] = [];
+    world.queue = async (queueName, message, opts) => {
+      const result = await originalQueue.call(world, queueName, message, opts);
+      if (
+        redelivered.length === 0 &&
+        queueName.startsWith('__wkf_workflow_') &&
+        (message as { runInput?: unknown }).runInput !== undefined
+      ) {
+        redelivered.push(
+          new Promise((resolve) => setTimeout(resolve, 500)).then(() =>
+            originalQueue.call(world, queueName, message, opts)
+          )
+        );
+      }
+      return result;
+    };
+
+    try {
+      const run = await start(inlineStepDuringRedeliveryWorkflow, [markerPath]);
+      await expect(run.returnValue).resolves.toBe('done');
+      expect(redelivered).toHaveLength(1);
+      await Promise.all(redelivered);
+    } finally {
+      world.queue = originalQueue;
+    }
+
+    const marker = await readFile(markerPath, 'utf8');
+    const executions = marker.split('\n').filter(Boolean);
+    expect(
+      executions,
+      'the inline step body must execute exactly once despite the redelivery'
+    ).toHaveLength(1);
+  }, 60_000);
+});

workbench/vitest/workflows/inline-step-ownership.ts

@@ -30,3 +30,27 @@ export async function inlineStepDuringHookResumeWorkflow(
   const [stepResult] = await Promise.all([slowMarkerStep(markerPath), hook]);
   return stepResult;
 }
+
+/**
+ * Step whose side effect (an appended marker line) happens at the TOP of the
+ * body, so the marker counts body entries rather than completed attempts.
+ */
+async function slowEntryMarkerStep(markerPath: string) {
+  'use step';
+  await appendFile(markerPath, 'entered\n');
+  await new Promise((resolve) => setTimeout(resolve, 1500));
+  return 'done';
+}
+
+/**
+ * Regression workflow for issue #3909: a step-only workflow runs its first
+ * delivery in turbo mode with a lazy inline step. A redelivery of the same
+ * start message while the body is still running (world-local's transport
+ * timeout, or any at-least-once duplicate) re-enters turbo at attempt 1 and
+ * must not execute the body a second time.
+ */
+export async function inlineStepDuringRedeliveryWorkflow(markerPath: string) {
+  'use workflow';
+
+  return await slowEntryMarkerStep(markerPath);
+}