Retry 5xx errors from workflow-server in step handler (#1011)

* Retry 5xx errors from workflow-server in step handler

Add `withServerErrorRetry` helper that retries world calls on 5xx errors
with exponential backoff (500ms, 1s, 2s ≈ 3.5s total). Applied to all
`world.events.create` calls in the step handler so transient
workflow-server errors don't consume step attempts.

If retries are exhausted, the error is thrown to the queue for
higher-level retry without burning a step attempt.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Address PR review comments on 5xx retry handling

- Fix misleading `maxAttempts` log field to `maxRetries` in withServerErrorRetry
- Update step-handler comment to accurately note that queue retries may
  still consume step attempts since step_started has already incremented
- Add unit tests for withServerErrorRetry (7 tests covering success,
  retry/backoff, exhaustion, and non-5xx passthrough)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Add tests for 429/5xx retry handling

Unit tests for withThrottleRetry and withServerErrorRetry helpers, plus
an e2e test that exercises the 5xx retry codepath during step execution
via run-scoped fault injection.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Wrap VQS errors in WorkflowAPIError and add retry to queueMessage

VQS throws its own error types (InternalServerError, ConsumerDiscoveryError,
ConsumerRegistryNotConfiguredError) that don't match WorkflowAPIError.is().
Wrapping them at the world-vercel boundary enables withServerErrorRetry in
queueMessage() to automatically retry transient queue failures.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Update changeset to include @workflow/world-vercel

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

* Remove queue retry logic per review feedback

Queue retrying will be handled natively by the @vercel/queue client
instead. Reverts VQS error wrapping and withServerErrorRetry in
queueMessage().

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: pranaygp

.changeset/retry-5xx-step-handler.md

@@ -0,0 +1,5 @@
+---
+"@workflow/core": patch
+---
+
+Retry 5xx errors from workflow-server in step handler to avoid consuming step attempts on transient infrastructure errors

packages/core/e2e/e2e.test.ts

@@ -889,6 +889,33 @@ describe('e2e', () => {
         expect(result.failed).toBe(true);
         expect(result.attempt).toBe(1);
       });
+
+      test(
+        'workflow completes despite transient 5xx on step_completed',
+        { timeout: 120_000 },
+        async () => {
+          const run = await start(
+            await e2e('serverError5xxRetryWorkflow'),
+            [42]
+          );
+          const result = await run.returnValue;
+
+          // Correct result proves workflow completed successfully
+          expect(result.result).toBe(84); // 42 * 2
+
+          // retryCount > 0 proves the fault injection actually triggered
+          expect(result.retryCount).toBe(2);
+
+          // attempt === 1 proves no step attempt was consumed by the 5xx retries
+          const { json: steps } = await cliInspectJson(
+            `steps --runId ${run.runId}`
+          );
+          const doWorkStep = steps.find((s: any) =>
+            s.stepName.includes('doWork')
+          );
+          expect(doWorkStep.attempt).toBe(1);
+        }
+      );
     });
 
     describe('catchability', () => {

packages/core/src/runtime/helpers.test.ts

@@ -1,5 +1,20 @@
-import { describe, expect, it } from 'vitest';
-import { getWorkflowQueueName } from './helpers';
+import { WorkflowAPIError } from '@workflow/errors';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
+import {
+  getWorkflowQueueName,
+  withServerErrorRetry,
+  withThrottleRetry,
+} from './helpers.js';
+
+// Mock the logger to suppress output during tests
+vi.mock('../logger.js', () => ({
+  runtimeLogger: {
+    warn: vi.fn(),
+    debug: vi.fn(),
+    info: vi.fn(),
+    error: vi.fn(),
+  },
+}));
 
 describe('getWorkflowQueueName', () => {
   it('should return a valid queue name for a simple workflow name', () => {
@@ -70,3 +85,227 @@ describe('getWorkflowQueueName', () => {
     expect(() => getWorkflowQueueName('')).toThrow('Invalid workflow name');
   });
 });
+
+describe('withServerErrorRetry', () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.clearAllMocks();
+  });
+
+  it('should return the result on success', async () => {
+    const fn = vi.fn().mockResolvedValue('ok');
+    const result = await withServerErrorRetry(fn);
+    expect(result).toBe('ok');
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+
+  it('should retry on 5xx WorkflowAPIError and succeed', async () => {
+    const fn = vi
+      .fn()
+      .mockRejectedValueOnce(
+        new WorkflowAPIError('Internal Server Error', { status: 500 })
+      )
+      .mockResolvedValueOnce('recovered');
+
+    const promise = withServerErrorRetry(fn);
+    await vi.advanceTimersByTimeAsync(500);
+    const result = await promise;
+
+    expect(result).toBe('recovered');
+    expect(fn).toHaveBeenCalledTimes(2);
+  });
+
+  it('should retry up to 3 times with exponential backoff', async () => {
+    const fn = vi
+      .fn()
+      .mockRejectedValueOnce(new WorkflowAPIError('error', { status: 502 }))
+      .mockRejectedValueOnce(new WorkflowAPIError('error', { status: 503 }))
+      .mockRejectedValueOnce(new WorkflowAPIError('error', { status: 500 }))
+      .mockResolvedValueOnce('finally');
+
+    const promise = withServerErrorRetry(fn);
+
+    // First retry after 500ms
+    await vi.advanceTimersByTimeAsync(500);
+    // Second retry after 1000ms
+    await vi.advanceTimersByTimeAsync(1000);
+    // Third retry after 2000ms
+    await vi.advanceTimersByTimeAsync(2000);
+
+    const result = await promise;
+    expect(result).toBe('finally');
+    expect(fn).toHaveBeenCalledTimes(4);
+  });
+
+  it('should throw after exhausting all retries', async () => {
+    const error = new WorkflowAPIError('server down', { status: 500 });
+    const fn = vi.fn().mockRejectedValue(error);
+
+    const promise = withServerErrorRetry(fn).catch((e) => e);
+
+    // Advance through all 3 retry delays
+    await vi.advanceTimersByTimeAsync(500);
+    await vi.advanceTimersByTimeAsync(1000);
+    await vi.advanceTimersByTimeAsync(2000);
+
+    const result = await promise;
+    expect(result).toBeInstanceOf(WorkflowAPIError);
+    expect(result.message).toBe('server down');
+    // 1 initial + 3 retries = 4 total calls
+    expect(fn).toHaveBeenCalledTimes(4);
+  });
+
+  it('should not retry non-5xx WorkflowAPIErrors', async () => {
+    const error = new WorkflowAPIError('Not Found', { status: 404 });
+    const fn = vi.fn().mockRejectedValue(error);
+
+    await expect(withServerErrorRetry(fn)).rejects.toThrow('Not Found');
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+
+  it('should not retry non-WorkflowAPIError errors', async () => {
+    const error = new Error('some other error');
+    const fn = vi.fn().mockRejectedValue(error);
+
+    await expect(withServerErrorRetry(fn)).rejects.toThrow('some other error');
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+
+  it('should not retry 429 errors (handled by withThrottleRetry)', async () => {
+    const error = new WorkflowAPIError('Too Many Requests', {
+      status: 429,
+      retryAfter: 5,
+    });
+    const fn = vi.fn().mockRejectedValue(error);
+
+    await expect(withServerErrorRetry(fn)).rejects.toThrow('Too Many Requests');
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+});
+
+describe('withThrottleRetry', () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.clearAllMocks();
+  });
+
+  it('should pass through the result on success', async () => {
+    const fn = vi.fn().mockResolvedValue(undefined);
+    const result = await withThrottleRetry(fn);
+    expect(result).toBeUndefined();
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+
+  it('should pass through { timeoutSeconds } returned by fn', async () => {
+    const fn = vi.fn().mockResolvedValue({ timeoutSeconds: 42 });
+    const result = await withThrottleRetry(fn);
+    expect(result).toEqual({ timeoutSeconds: 42 });
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+
+  it('should re-throw non-429 errors including 5xx', async () => {
+    const error = new WorkflowAPIError('Internal Server Error', {
+      status: 500,
+    });
+    const fn = vi.fn().mockRejectedValue(error);
+
+    await expect(withThrottleRetry(fn)).rejects.toThrow(
+      'Internal Server Error'
+    );
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+
+  it('should re-throw non-WorkflowAPIError errors', async () => {
+    const error = new Error('random failure');
+    const fn = vi.fn().mockRejectedValue(error);
+
+    await expect(withThrottleRetry(fn)).rejects.toThrow('random failure');
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+
+  it('should wait in-process and retry once for short retryAfter (<10s)', async () => {
+    const fn = vi
+      .fn()
+      .mockRejectedValueOnce(
+        new WorkflowAPIError('Throttled', { status: 429, retryAfter: 5 })
+      )
+      .mockResolvedValueOnce(undefined);
+
+    const promise = withThrottleRetry(fn);
+    // Advance past the 5s wait
+    await vi.advanceTimersByTimeAsync(5000);
+    const result = await promise;
+
+    expect(result).toBeUndefined();
+    expect(fn).toHaveBeenCalledTimes(2);
+  });
+
+  it('should defer to queue when both attempts are throttled (double 429)', async () => {
+    const fn = vi
+      .fn()
+      .mockRejectedValueOnce(
+        new WorkflowAPIError('Throttled', { status: 429, retryAfter: 3 })
+      )
+      .mockRejectedValueOnce(
+        new WorkflowAPIError('Throttled again', { status: 429, retryAfter: 7 })
+      );
+
+    const promise = withThrottleRetry(fn);
+    // Advance past the 3s in-process wait
+    await vi.advanceTimersByTimeAsync(3000);
+    const result = await promise;
+
+    expect(result).toEqual({ timeoutSeconds: 7 });
+    expect(fn).toHaveBeenCalledTimes(2);
+  });
+
+  it('should re-throw non-429 error on retry failure', async () => {
+    const fn = vi
+      .fn()
+      .mockRejectedValueOnce(
+        new WorkflowAPIError('Throttled', { status: 429, retryAfter: 2 })
+      )
+      .mockRejectedValueOnce(new Error('connection lost'));
+
+    // Capture the rejection early to prevent unhandled rejection warning
+    const promise = withThrottleRetry(fn).catch((e) => e);
+    await vi.advanceTimersByTimeAsync(2000);
+
+    const result = await promise;
+    expect(result).toBeInstanceOf(Error);
+    expect(result.message).toBe('connection lost');
+    expect(fn).toHaveBeenCalledTimes(2);
+  });
+
+  it('should defer to queue immediately for long retryAfter (>=10s)', async () => {
+    const fn = vi
+      .fn()
+      .mockRejectedValueOnce(
+        new WorkflowAPIError('Throttled', { status: 429, retryAfter: 15 })
+      );
+
+    const result = await withThrottleRetry(fn);
+
+    expect(result).toEqual({ timeoutSeconds: 15 });
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+
+  it('should default to 30s (defer to queue) when no retryAfter is provided', async () => {
+    const error = new WorkflowAPIError('Throttled', { status: 429 });
+    // retryAfter is undefined, so it defaults to 30 (>=10 → defer)
+    const fn = vi.fn().mockRejectedValue(error);
+
+    const result = await withThrottleRetry(fn);
+
+    expect(result).toEqual({ timeoutSeconds: 30 });
+    expect(fn).toHaveBeenCalledTimes(1);
+  });
+});

packages/core/src/runtime/helpers.ts

@@ -468,3 +468,43 @@ export async function withThrottleRetry(
     throw err;
   }
 }
+
+/**
+ * Retries a function when it throws a 5xx WorkflowAPIError.
+ * Used to handle transient workflow-server errors without consuming step attempts.
+ *
+ * Retries up to 3 times with exponential backoff (500ms, 1s, 2s ≈ 3.5s total).
+ * If all retries fail, the original error is re-thrown.
+ */
+export async function withServerErrorRetry<T>(
+  fn: () => Promise<T>
+): Promise<T> {
+  const delays = [500, 1000, 2000];
+  for (let attempt = 0; attempt <= delays.length; attempt++) {
+    try {
+      return await fn();
+    } catch (err) {
+      if (
+        WorkflowAPIError.is(err) &&
+        err.status !== undefined &&
+        err.status >= 500 &&
+        attempt < delays.length
+      ) {
+        runtimeLogger.warn(
+          'Server error (5xx) from workflow-server, retrying in-process',
+          {
+            status: err.status,
+            attempt: attempt + 1,
+            maxRetries: delays.length,
+            nextDelayMs: delays[attempt],
+            url: err.url,
+          }
+        );
+        await new Promise((resolve) => setTimeout(resolve, delays[attempt]));
+        continue;
+      }
+      throw err;
+    }
+  }
+  throw new Error('withServerErrorRetry: unreachable');
+}