[core] Enforce maxRetries for steps that time out

.changeset/step-maxretries-timeout.md

@@ -0,0 +1,5 @@
+---
+'@workflow/core': patch
+---
+
+Enforce `maxRetries` for inline and backgrounded steps that time out

packages/core/src/runtime.ts

@@ -61,7 +61,11 @@ import {
   ReplayBudget,
 } from './runtime/replay-budget.js';
 import { runIdCreatedAt } from './runtime/run-id-time.js';
-import { executeStep } from './runtime/step-executor.js';
+import {
+  DEFAULT_STEP_MAX_RETRIES,
+  executeStep,
+} from './runtime/step-executor.js';
+import { getStepFunction } from './private.js';
 import { computeStepLatencyTracking } from './runtime/step-latency.js';
 import {
   backstopIdempotencyKey,
@@ -246,6 +250,29 @@ function hasRecordedTerminalRunEvent(events: Event[], runId: string): boolean {
   return true;
 }
 
+/**
+ * Number of `step_started` events already recorded for a step, used as the
+ * authoritative attempt count for the inline retry ceiling. Each real attempt
+ * writes exactly one `step_started` (the atomic create-claim / single-flight
+ * prevents concurrent double-starts from inflating this), so the count equals
+ * the number of attempts that have begun.
+ */
+function countStepStartedEvents(
+  events: Event[] | null | undefined,
+  stepId: string
+): number {
+  if (!events) {
+    return 0;
+  }
+  let count = 0;
+  for (const e of events) {
+    if (e.eventType === 'step_started' && e.correlationId === stepId) {
+      count++;
+    }
+  }
+  return count;
+}
+
 /**
  * The lineage root of a loaded run: its `$rootRunId` attribute, or its own id
  * when it is itself a root.
@@ -720,6 +747,38 @@ export function workflowEntrypoint(
                       const bgStartedAt = bgRun.startedAt
                         ? +bgRun.startedAt
                         : Date.now();
+
+                      // Retry ceiling for a backgrounded step. `metadata.attempt`
+                      // (the queue delivery count) is a cheap upper bound, but it
+                      // over-counts: a ThrottleError / TooEarlyError — or any
+                      // redelivery that never ran the body — still advances it, so
+                      // trusting it directly could fail a step as "exceeded max
+                      // retries" before the body ever ran (a user-visible
+                      // regression under transient backend pressure). Use it only
+                      // as a fast gate: while it is at or under the ceiling the
+                      // step cannot be exhausted, so proceed without touching the
+                      // log. Only once it crosses the ceiling do we load the full
+                      // event log and derive the authoritative attempt from the
+                      // recorded `step_started` count — the count only real
+                      // attempts write, so throttle/too-early redeliveries are
+                      // excluded. This still bounds timeouts, which write no error
+                      // for the post-body guard to catch. The load also primes the
+                      // replay's `cachedEvents`/`eventsCursor` (the post-step
+                      // continuation below refreshes them once the step's terminal
+                      // event lands).
+                      let bgAuthoritativeAttempt = metadata.attempt;
+                      const bgMaxRetries =
+                        getStepFunction(incomingStepName)?.maxRetries ??
+                        DEFAULT_STEP_MAX_RETRIES;
+                      if (metadata.attempt > bgMaxRetries + 1) {
+                        const loaded = await loadWorkflowRunEvents(runId);
+                        cachedEvents = loaded.events;
+                        eventsCursor = loaded.cursor;
+                        bgAuthoritativeAttempt =
+                          countStepStartedEvents(cachedEvents, incomingStepId) +
+                          1;
+                      }
+
                       // Pause the replay budget while the step body runs —
                       // step duration is bounded by the platform's function
                       // maxDuration, not by the replay timeout. See the
@@ -751,6 +810,10 @@ export function workflowEntrypoint(
                               stepId: incomingStepId,
                               stepName: incomingStepName,
                               runSpecVersion: bgRun.specVersion,
+                              // Retry ceiling: the queue delivery count as a fast
+                              // gate, verified against the recorded step_started
+                              // count once it crosses the ceiling (see above).
+                              authoritativeAttempt: bgAuthoritativeAttempt,
                             })
                         );
                       } finally {
@@ -2242,6 +2305,25 @@ export function workflowEntrypoint(
                                 stepId: s.correlationId,
                                 stepName: s.stepName,
                                 runSpecVersion: workflowRun.specVersion,
+                                // Attempt number = prior step_started count + 1
+                                // (this execution's start). A lazy step is
+                                // brand-new by construction (it enters the batch
+                                // only when it has no step_created yet), so it
+                                // has zero prior starts and is always attempt 1 —
+                                // skip the log scan entirely. Only an
+                                // owned-recovery re-run (this message re-executing
+                                // a step it crashed/timed out on) can have prior
+                                // starts, and that path is uncommon, so reserve
+                                // the O(n) scan for it rather than walking the
+                                // growing log for every inline step (which would
+                                // be O(n²) across a long sequential workflow).
+                                authoritativeAttempt:
+                                  s.lazyStepInput !== undefined
+                                    ? 1
+                                    : countStepStartedEvents(
+                                        cachedEvents,
+                                        s.correlationId
+                                      ) + 1,
                                 // Lazy inline start: send the deferred step's
                                 // input on step_started so the world creates
                                 // the step on the fly. Absent for

packages/core/src/runtime/step-executor.test.ts

@@ -0,0 +1,166 @@
+import { mkdtempSync } from 'node:fs';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import type { Event, World } from '@workflow/world';
+import { SPEC_VERSION_CURRENT } from '@workflow/world';
+import { createWorld } from '@workflow/world-local';
+import { afterEach, describe, expect, it } from 'vitest';
+import { registerStepFunction } from '../private.js';
+import { dehydrateStepArguments } from '../serialization.js';
+import { executeStep } from './step-executor.js';
+
+// The retry ceiling (`authoritativeAttempt`) is what bounds a step that keeps
+// timing out: a timeout hard-kills the body without writing any error, so the
+// error-based guards never fire. These tests assert the ceiling is enforced
+// BEFORE the body runs, and only once the attempt number actually exceeds
+// maxRetries + 1.
+
+const MAX_RETRIES = 3; // maxRetries + 1 = 4 total attempts allowed
+
+let counter = 0;
+function uniqueStepName(): string {
+  counter += 1;
+  return `step//./step-executor-test//timeoutStep${counter}`;
+}
+
+async function setupRunningStep(opts: {
+  world: World;
+  stepName: string;
+  onBody: () => void;
+}): Promise<{ runId: string; stepId: string }> {
+  const { world, stepName, onBody } = opts;
+  const runInput = await dehydrateStepArguments([], 'run', undefined);
+  const created = await world.events.create(null, {
+    eventType: 'run_created',
+    specVersion: SPEC_VERSION_CURRENT,
+    eventData: {
+      deploymentId: 'dpl_test',
+      workflowName: 'wf',
+      input: runInput,
+    },
+  });
+  const runId = created.run!.runId;
+  await world.events.create(runId, {
+    eventType: 'run_started',
+    specVersion: SPEC_VERSION_CURRENT,
+    eventData: {},
+  } as never);
+
+  const stepId = 'step_timeout_1';
+  const stepInput = await dehydrateStepArguments([], runId, undefined);
+  await world.events.create(runId, {
+    eventType: 'step_created',
+    specVersion: SPEC_VERSION_CURRENT,
+    correlationId: stepId,
+    eventData: { stepName, input: stepInput },
+  });
+
+  const stepFn = Object.assign(
+    async () => {
+      onBody();
+      return 'ok';
+    },
+    { maxRetries: MAX_RETRIES }
+  );
+  registerStepFunction(stepName, stepFn);
+
+  return { runId, stepId };
+}
+
+function makeWorld(): World {
+  const dataDir = mkdtempSync(join(tmpdir(), 'wf-step-executor-'));
+  return createWorld({ dataDir, tag: `t${counter}` });
+}
+
+async function eventsFor(
+  world: World,
+  runId: string,
+  stepId: string,
+  eventType: Event['eventType']
+): Promise<Event[]> {
+  const { data } = await world.events.list({ runId });
+  return data.filter(
+    (e) => e.eventType === eventType && e.correlationId === stepId
+  );
+}
+
+describe('executeStep — retry ceiling (authoritativeAttempt)', () => {
+  afterEach(() => {
+    counter += 1;
+  });
+
+  it('fails the step WITHOUT running the body once the attempt exceeds maxRetries + 1', async () => {
+    const world = makeWorld();
+    const stepName = uniqueStepName();
+    let bodyRuns = 0;
+    const { runId, stepId } = await setupRunningStep({
+      world,
+      stepName,
+      onBody: () => {
+        bodyRuns += 1;
+      },
+    });
+
+    const result = await executeStep({
+      world,
+      workflowRunId: runId,
+      workflowName: 'wf',
+      workflowStartedAt: Date.now(),
+      stepId,
+      stepName,
+      // Attempt maxRetries + 2 — one past the last allowed retry. This is the
+      // delivery a timed-out step would land on with nothing left to try.
+      authoritativeAttempt: MAX_RETRIES + 2,
+    });
+
+    expect(result.type).toBe('failed');
+    // The body must NOT run — retries are already exhausted.
+    expect(bodyRuns).toBe(0);
+
+    // The ceiling fires BEFORE the start block, so no new step_started is
+    // written for the rejected attempt; the step goes straight to failed.
+    const started = await eventsFor(world, runId, stepId, 'step_started');
+    expect(started).toHaveLength(0);
+    const failures = await eventsFor(world, runId, stepId, 'step_failed');
+    expect(failures).toHaveLength(1);
+  });
+
+  it('permits (does not pre-empt) the final allowed attempt (maxRetries + 1)', async () => {
+    const world = makeWorld();
+    const stepName = uniqueStepName();
+    let bodyRuns = 0;
+    const { runId, stepId } = await setupRunningStep({
+      world,
+      stepName,
+      onBody: () => {
+        bodyRuns += 1;
+      },
+    });
+
+    const result = await executeStep({
+      world,
+      workflowRunId: runId,
+      workflowName: 'wf',
+      workflowStartedAt: Date.now(),
+      stepId,
+      stepName,
+      // maxRetries + 1 is the last permitted attempt: the ceiling must let it
+      // proceed into normal execution rather than pre-emptively failing it.
+      authoritativeAttempt: MAX_RETRIES + 1,
+    });
+
+    // It got past the ceiling: the step was started (entered normal execution)
+    // and was NOT failed by the retry ceiling.
+    void bodyRuns;
+    expect(result.type).not.toBe('failed');
+    const started = await eventsFor(world, runId, stepId, 'step_started');
+    expect(started).toHaveLength(1);
+    const ceilingFailures = await eventsFor(
+      world,
+      runId,
+      stepId,
+      'step_failed'
+    );
+    expect(ceilingFailures).toHaveLength(0);
+  });
+});

packages/core/src/runtime/step-executor.ts

@@ -51,7 +51,7 @@ import {
 } from './step-latency.js';
 import { safeWaitUntil } from './wait-until.js';
 
-const DEFAULT_STEP_MAX_RETRIES = 3;
+export const DEFAULT_STEP_MAX_RETRIES = 3;
 
 /**
  * Extract the inline delta from a step-terminal `events.create` result,
@@ -180,6 +180,17 @@ export interface StepExecutorParams {
    * runtime/step-latency.ts.
    */
   latencyTracking?: StepLatencyTracking;
+  /**
+   * Authoritative attempt number for this execution, used to bound retries
+   * against `maxRetries` BEFORE the body runs. In order to also catch
+   * step timeouts (which we can't have catch handlers for), we determine
+   * the attempt count based as follows:
+   * - Inline (combined handler): the number of `step_started` events already
+   *   in the event log for this step, plus one for the attempt about to run.
+   * - Background (queue-dispatched): the queue delivery count
+   *   (`metadata.attempt`), which increments on every redelivery.
+   */
+  authoritativeAttempt?: number;
 }
 
 /**
@@ -351,6 +362,70 @@ export async function executeStep(
       ...Attribute.StepMaxRetries(maxRetries),
     });
 
+    // maxRetries enforced before starting another attempt. Timeouts can kill
+    // a step execution without any way to catch the error, so the post-body/error
+    // guards below might miss it. Hence, we use `authoritativeAttempt` count
+    // as an additional guard based on step_started count or queue delivery count
+    // on backgrounded steps.
+    // Thrown-error exhaustion still terminates one attempt earlier via the post-body
+    // guard (with the thrown error as cause), and this check does not affect that.
+    if (
+      params.authoritativeAttempt !== undefined &&
+      params.authoritativeAttempt > maxRetries + 1
+    ) {
+      const retryCount = maxRetries;
+      const errorMessage = `Step "${stepName}" exceeded max retries (${retryCount} ${pluralize('retry', 'retries', retryCount)})`;
+      stepLogger.error('Step exceeded max retries', {
+        workflowRunId,
+        stepName,
+        stepId,
+        attempt: params.authoritativeAttempt,
+        maxRetries,
+      });
+      try {
+        await world.events.create(workflowRunId, {
+          eventType: 'step_failed',
+          specVersion: SPEC_VERSION_CURRENT,
+          correlationId: stepId,
+          eventData: {
+            stepName,
+            error: await dehydrateStepError(
+              new FatalError(errorMessage),
+              workflowRunId,
+              await getEncryptionKey(),
+              [],
+              globalThis,
+              compression
+            ),
+          },
+        });
+      } catch (err) {
+        if (EntityConflictError.is(err)) {
+          // Step already reached a terminal state (a concurrent handler or an
+          // earlier delivery failed/completed it) — nothing to do.
+          runtimeLogger.info(
+            'Tried failing step for exceeded retries, but step has already finished.',
+            {
+              workflowRunId,
+              stepId,
+              stepName,
+              message: err instanceof Error ? err.message : String(err),
+            }
+          );
+          return { type: 'skipped' };
+        }
+        if (RunExpiredError.is(err)) {
+          return { type: 'gone' };
+        }
+        throw err;
+      }
+      span?.setAttributes({
+        ...Attribute.StepStatus('failed'),
+        ...Attribute.StepRetryExhausted(true),
+      });
+      return { type: 'failed' };
+    }
+
     // Maps a `step_started` rejection to a terminal StepExecutionResult,
     // shared by the await path (below) and the optimistic-start reconciliation.
     // Returns undefined when the error is not one we translate (caller rethrows).