[core] Re-dispatch a pending step whose dispatch ended without a terminal event

.changeset/wild-pandas-jam.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Re-dispatch a pending step whose dispatch ended without a terminal event, so a run is no longer stranded with a step nothing will execute

.github/workflows/event-log-race-repro.yml

@@ -62,11 +62,14 @@ jobs:
     name: Event Log Race Repro
     runs-on: ubuntu-latest
     # Sized for the harness' default scale: its own test timeout is
-    # `budget_ms + run_timeout_ms + 60s` (~17 min at the defaults), and the rest is
-    # checkout, build, the deployment wait, and rendering the summary. A soak
-    # dispatch that raises `budget_ms` has to raise this too, or the runner kills
-    # the job before the summary is written.
-    timeout-minutes: 25
+    # `budget_ms + run_timeout_ms + 60s` (~30 min at the defaults, where
+    # `run_timeout_ms` is derived from the runtime's inline-ownership lease so a
+    # recovering run is not misreported as stuck), and the rest is checkout,
+    # build, the deployment wait, and rendering the summary. Only a run that
+    # actually needs recovery spends that deadline; healthy attempts finish in
+    # tens of seconds. A soak dispatch that raises `budget_ms` has to raise this
+    # too, or the runner kills the job before the summary is written.
+    timeout-minutes: 40
     if: ${{ github.event_name == 'workflow_dispatch' || contains(github.event.pull_request.labels.*.name, 'event-log-race-repro') }}
     permissions:
       contents: read

docs/content/docs/v5/configuration/runtime-tuning.mdx

@@ -166,6 +166,16 @@ For example, a workflow can run a 10-minute inline step even with `WORKFLOW_REPL
 - How long after an inline step's latest `step_started` other invocations assume its owner may still be executing the body. Within the lease they defer the step's backstop message; past it they enqueue immediately.
 - Raise this on self-hosted multi-instance deployments whose inline steps run longer than the default (the default is sized for Vercel's function duration ceiling).
 
+### `WORKFLOW_STEP_DISPATCH_WATCHDOG_SECONDS`
+
+- Default: `60`
+- Clamp: `10` to `900`
+- How long a step may sit created but never started before the runtime re-dispatches it. A queued step message is deduplicated on a key derived from the step's identity, and a queue holds that claim on its own schedule rather than releasing it when the message finishes. So a dispatch that ends without the step reaching a terminal event cannot be revived by sending the same key again, and the run keeps replaying with nothing to execute. Past this interval the re-dispatch uses a key the queue has not seen, and the runtime arms a timer on the boundary so the check happens even when nothing else wakes the run.
+- This does not compensate for a queue dropping messages. An undelivered or unacknowledged message is redelivered by the queue itself, and a redundant send being deduplicated is harmless while a dispatch is still in flight. What the watchdog re-opens is a dispatch with nothing outstanding: the message was acknowledged, or collapsed into a claim whose message has already finished, without a terminal event for the step.
+- Re-dispatch is at-least-once: if the original message was only slow, both deliveries execute the step body and the loser's result is discarded. Raise this if your queue's dispatch-to-start latency approaches the default.
+- A step that has already started gets the same treatment on a later deadline: it is presumed alive for the inline ownership lease above, and only past the lease does its re-dispatch move to a fresh key. Once past that point, the interval here is also how often the re-dispatch is retried.
+- Steps waiting on a scheduled retry (`step_retrying`) are left alone, since that retry is already queued with its own backoff.
+
 ## Workflow VM engine
 
 ### `WORKFLOW_VM`

packages/core/e2e/event-log-race-repro.test.ts

@@ -10,6 +10,10 @@ import {
   start as rawStart,
   resumeHook,
 } from '../src/runtime';
+import {
+  getInlineOwnershipLeaseSeconds,
+  getStepDispatchWatchdogSeconds,
+} from '../src/runtime/constants';
 import { getWorkflowMetadata, setupWorld, trackRun } from './utils';
 
 /**
@@ -158,6 +162,33 @@ function envBoolean(name: string, fallback: boolean) {
 // `workflow_dispatch` inputs straight through and `envNumber` treats an unset or
 // empty variable as absent, so a blank input lands on the value below rather
 // than on a second default maintained in YAML.
+/**
+ * How long a run may take before the harness calls it `stuck`.
+ *
+ * A healthy attempt finishes in tens of seconds, so this is not a latency
+ * budget: it is the point past which a run is declared unrecoverable. That
+ * makes it meaningless to set it below the runtime's own longest recovery
+ * deadline. A step whose owning invocation disappears mid-body is presumed
+ * alive for the inline-ownership lease, and its re-dispatch reaches the queue
+ * one watchdog interval later at worst, so anything shorter reports a run that
+ * is on its way back as permanently stranded.
+ *
+ * Both deadlines are read from this process's environment, which matches the
+ * deployment under test only when neither side overrides them. Overriding the
+ * lease or the watchdog on the deployment means setting
+ * `EVENT_LOG_RACE_REPRO_RUN_TIMEOUT_MS` here to match.
+ */
+function defaultRunTimeoutMs(): number {
+  const recoveryMs =
+    (getInlineOwnershipLeaseSeconds() + getStepDispatchWatchdogSeconds()) *
+    1000;
+  // The recovery deadline is measured from the lost step's start, not from the
+  // run's, so the slack has to cover a run that gets that far in before it
+  // loses a step, plus delivering the recovered dispatch and finishing the
+  // remaining rounds.
+  return recoveryMs + 3 * 60_000;
+}
+
 const config: ReproConfig = {
   stepStormAttempts: envNumber('EVENT_LOG_RACE_REPRO_STEP_STORM_ATTEMPTS', 6),
   hookStormAttempts: envNumber('EVENT_LOG_RACE_REPRO_HOOK_STORM_ATTEMPTS', 6),
@@ -173,7 +204,10 @@ const config: ReproConfig = {
   // absorb the worst case of one in-flight attempt draining its full
   // `runTimeoutMs` after the budget ends (see `testTimeoutMs` below).
   budgetMs: envNumber('EVENT_LOG_RACE_REPRO_BUDGET_MS', 12 * 60_000),
-  runTimeoutMs: envNumber('EVENT_LOG_RACE_REPRO_RUN_TIMEOUT_MS', 240_000),
+  runTimeoutMs: envNumber(
+    'EVENT_LOG_RACE_REPRO_RUN_TIMEOUT_MS',
+    defaultRunTimeoutMs()
+  ),
   hookTimeoutMs: envNumber('EVENT_LOG_RACE_REPRO_HOOK_TIMEOUT_MS', 60_000),
   rounds: envNumber('EVENT_LOG_RACE_REPRO_ROUNDS', 6),
   width: envNumber('EVENT_LOG_RACE_REPRO_WIDTH', 8),
@@ -915,7 +949,6 @@ const plannedAttempts =
   config.hookSleepAttempts;
 let overallDeadline = Number.POSITIVE_INFINITY;
 let launchDeadline = Number.POSITIVE_INFINITY;
-let remainingPlanned = plannedAttempts;
 let budgetExhausted = false;
 let lastCheckpointAt = 0;
 
@@ -951,35 +984,60 @@ function recordResult(result: ReproRunResult) {
   }
 }
 
-async function runScenario(
-  attempts: number,
-  concurrency: number,
-  run: (attempt: number) => Promise<ReproRunResult>
-) {
-  if (attempts <= 0) {
-    return [];
+interface ScenarioPlan {
+  scenario: Scenario;
+  attempts: number;
+  run: (attempt: number) => Promise<ReproRunResult>;
+}
+
+/**
+ * Every planned attempt in one launch order, with each scenario's attempts
+ * spread across the whole order rather than grouped into a contiguous block.
+ *
+ * Blocks make the launch budget positional. One attempt that spends its full
+ * `runTimeoutMs` holds its block open past the budget, and every scenario
+ * behind it reports zero runs: `hook-storm` is the production shape and
+ * `hook-sleep` is the calibration control, so a truncated run loses exactly the
+ * parts of the result that carry the most meaning. Interleaved, a truncated run
+ * is proportionally short in every scenario instead.
+ *
+ * Cross-run concurrency is unchanged. One `mapLimit` over this order holds
+ * `config.concurrency` attempts in flight no matter which scenarios they belong
+ * to, and the race each attempt reproduces is between replays *within* its own
+ * run, so which scenarios its neighbours are running does not enter into it.
+ */
+function interleaveAttempts(plans: ScenarioPlan[]) {
+  const queues = plans
+    .filter((plan) => plan.attempts > 0)
+    .map((plan) => ({ plan, issued: 0 }));
+  const order: {
+    scenario: Scenario;
+    attempt: number;
+    run: () => Promise<ReproRunResult>;
+  }[] = [];
+  for (;;) {
+    // Whichever scenario is furthest behind its share of the order goes next,
+    // so every prefix of the order tracks the planned proportions.
+    let next: (typeof queues)[number] | undefined;
+    for (const queue of queues) {
+      if (queue.issued >= queue.plan.attempts) {
+        continue;
+      }
+      if (
+        next === undefined ||
+        queue.issued / queue.plan.attempts < next.issued / next.plan.attempts
+      ) {
+        next = queue;
+      }
+    }
+    if (next === undefined) {
+      return order;
+    }
+    next.issued += 1;
+    const { scenario, run } = next.plan;
+    const attempt = next.issued;
+    order.push({ scenario, attempt, run: () => run(attempt) });
   }
-  // Each scenario gets the share of the remaining budget its remaining planned
-  // attempts represent, so a truncated run still carries data for every
-  // scenario — including the `hook-sleep` control, which runs last and would
-  // otherwise be the first thing a single global deadline dropped. A scenario
-  // that finishes under its slice hands the surplus to the next one, since the
-  // slice is recomputed from the wall-clock left until `overallDeadline`.
-  const now = Date.now();
-  launchDeadline = Math.min(
-    overallDeadline,
-    now + ((overallDeadline - now) * attempts) / remainingPlanned
-  );
-  remainingPlanned -= attempts;
-  const attemptNumbers = Array.from(
-    { length: attempts },
-    (_, index) => index + 1
-  );
-  return await mapLimit(attemptNumbers, concurrency, async (attempt) => {
-    const result = await run(attempt);
-    recordResult(result);
-    return result;
-  });
 }
 
 // Derived from the launch budget, not from the attempt count: the budget is
@@ -1043,25 +1101,33 @@ describe('event log race repro', () => {
     { timeout: testTimeoutMs },
     async () => {
       overallDeadline = Date.now() + config.budgetMs;
+      launchDeadline = overallDeadline;
       // Written up front so a kill before the first checkpoint still produces a
       // file the renderer can report against, rather than nothing at all.
       writeResults(collected, false);
 
-      await runScenario(
-        config.stepStormAttempts,
-        config.concurrency,
-        runStepStormAttempt
-      );
-      await runScenario(
-        config.hookStormAttempts,
-        config.concurrency,
-        runHookStormAttempt
-      );
-      await runScenario(
-        config.hookSleepAttempts,
-        config.concurrency,
-        runHookSleepAttempt
-      );
+      const order = interleaveAttempts([
+        {
+          scenario: 'step-storm',
+          attempts: config.stepStormAttempts,
+          run: runStepStormAttempt,
+        },
+        {
+          scenario: 'hook-storm',
+          attempts: config.hookStormAttempts,
+          run: runHookStormAttempt,
+        },
+        {
+          scenario: 'hook-sleep',
+          attempts: config.hookSleepAttempts,
+          run: runHookSleepAttempt,
+        },
+      ]);
+      await mapLimit(order, config.concurrency, async (item) => {
+        const result = await item.run();
+        recordResult(result);
+        return result;
+      });
 
       const results = collected;
       // A budget-exhausted run launched fewer than `plannedAttempts` attempts,

packages/core/src/global.ts

@@ -11,6 +11,14 @@ export interface StepInvocationQueueItem {
   closureVars?: Record<string, Serializable>;
   thisVal?: Serializable;
   hasCreatedEvent?: boolean;
+  /**
+   * `createdAt` (ms) of the step's durable `step_created`, when the replay
+   * observed one. Anchors the re-dispatch watchdog for a step that never
+   * started (runtime/step-dispatch.ts). Undefined for a step this suspension
+   * is creating right now, and on worlds whose events carry no usable
+   * timestamp — both keep the plain correlation-ID dispatch key.
+   */
+  createdEventAt?: number;
   /**
    * Inline step ownership, derived from the step's LATEST `step_started`
    * during replay: the queue message ID stamped by the invocation running

packages/core/src/runtime.ts

@@ -93,6 +93,10 @@ import {
 } from './runtime/replay-budget.js';
 import { ReplayRecoveryReporter } from './runtime/replay-recovery-reporter.js';
 import { runIdCreatedAt } from './runtime/run-id-time.js';
+import {
+  getStepDispatchWake,
+  stepDispatchEpoch,
+} from './runtime/step-dispatch.js';
 import {
   DEFAULT_STEP_MAX_RETRIES,
   executeStep,
@@ -3304,6 +3308,11 @@ export function workflowEntrypoint(
                         const dispatchNowMs = Date.now();
                         const ownedRecoverySteps: StepInvocationQueueItem[] =
                           [];
+                        // Steps whose step-execution message went out this
+                        // suspension, here or from the suspension handler's
+                        // parallel dispatch, for the re-dispatch watchdog's
+                        // boundary wake below.
+                        const dispatchedSteps: StepInvocationQueueItem[] = [];
                         let backstopWakesArmed = 0;
                         for (const step of pendingSteps) {
                           if (inlineCorrelationIds.has(step.correlationId)) {
@@ -3321,6 +3330,11 @@ export function workflowEntrypoint(
                               step.correlationId
                             )
                           ) {
+                            // Its message can still be lost, and this is the
+                            // only suspension that will ever see the step
+                            // without something else already waking the run,
+                            // so it still counts for the boundary wake.
+                            dispatchedSteps.push(step);
                             continue;
                           }
                           const ownershipActive =
@@ -3372,6 +3386,7 @@ export function workflowEntrypoint(
                             );
                             continue;
                           }
+                          dispatchedSteps.push(step);
                           dispatches.push(
                             queueMessage(
                               world,
@@ -3384,21 +3399,55 @@ export function workflowEntrypoint(
                                 requestedAt: new Date(),
                               },
                               {
-                                // Step-identity-scoped: dedupes against every
-                                // other dispatch of THIS step (concurrent
-                                // handlers, crash-recovery re-dispatch, the
-                                // suspension handler's resilient publish)
-                                // without absorbing a dispatch of a different
-                                // step under a reassigned correlation id —
-                                // see stepDispatchIdempotencyKey.
+                                // Step-identity-scoped, and epoch-scoped once
+                                // this step's dispatch is presumed lost: a
+                                // watchdog interval without a first start, or
+                                // an expired ownership lease with no terminal
+                                // event. A key claim outlives the message sent
+                                // under it, so a dispatch that stopped short of
+                                // a terminal event can only be retried under a
+                                // key the queue has not seen. Within an epoch
+                                // every replay
+                                // derives the same key, so concurrent wake
+                                // replays still collapse to one message. See
+                                // stepDispatchIdempotencyKey and
+                                // runtime/step-dispatch.ts.
                                 idempotencyKey: stepDispatchIdempotencyKey(
                                   step.correlationId,
-                                  step.stepName
+                                  step.stepName,
+                                  stepDispatchEpoch(step, dispatchNowMs)
                                 ),
                               }
                             )
                           );
                         }
+                        // Nothing else wakes a run whose only outstanding work
+                        // is a step dispatch that was lost, so the watchdog
+                        // needs its own timer: one delayed run continuation at
+                        // the earliest boundary among the steps dispatched
+                        // here. Deduped on that boundary, so repeated
+                        // suspensions within an interval arm it once.
+                        const dispatchWake = getStepDispatchWake(
+                          dispatchedSteps,
+                          dispatchNowMs
+                        );
+                        if (dispatchWake) {
+                          dispatches.push(
+                            queueMessage(
+                              world,
+                              getWorkflowQueueName(workflowName, namespace),
+                              {
+                                runId,
+                                traceCarrier,
+                                requestedAt: new Date(),
+                              },
+                              {
+                                delaySeconds: dispatchWake.delaySeconds,
+                                idempotencyKey: dispatchWake.idempotencyKey,
+                              }
+                            )
+                          );
+                        }
                         if (suspensionResult.waitTimeout) {
                           dispatches.push(
                             queueMessage(