[core] Handle 429 and 500 errors from worlds in runtime

.changeset/throttle-retry-handling.md

@@ -0,0 +1,8 @@
+---
+"@workflow/errors": patch
+"@workflow/world": patch
+"@workflow/world-vercel": patch
+"@workflow/core": patch
+---
+
+Add 429 throttle retry handling and 500 server error retry with exponential backoff to the workflow and step runtimes

packages/core/src/runtime/helpers.ts

@@ -1,3 +1,4 @@
+import { WorkflowAPIError } from '@workflow/errors';
 import type {
   Event,
   HealthCheckPayload,
@@ -6,6 +7,7 @@ import type {
 } from '@workflow/world';
 import { HealthCheckPayloadSchema } from '@workflow/world';
 import { monotonicFactory } from 'ulid';
+import { runtimeLogger } from '../logger.js';
 import * as Attribute from '../telemetry/semantic-conventions.js';
 import { getSpanKind, trace } from '../telemetry.js';
 import { getWorld } from './world.js';
@@ -17,7 +19,7 @@ const DEFAULT_HEALTH_CHECK_TIMEOUT = 30_000;
  * Pattern for safe workflow names. Only allows alphanumeric characters,
  * underscores, hyphens, dots, and forward slashes (for namespaced workflows).
  */
-const SAFE_WORKFLOW_NAME_PATTERN = /^[a-zA-Z0-9_\-.\/]+$/;
+const SAFE_WORKFLOW_NAME_PATTERN = /^[a-zA-Z0-9_\-./]+$/;
 
 /**
  * Validates a workflow name and returns the corresponding queue name.
@@ -398,3 +400,70 @@ export function getQueueOverhead(message: { requestedAt?: Date }) {
     return;
   }
 }
+
+/**
+ * Wraps a queue handler with HTTP 429 throttle retry logic.
+ * - retryAfter < 10s: waits in-process via setTimeout, then retries once
+ * - retryAfter >= 10s: returns { timeoutSeconds } to defer to the queue
+ *
+ * Safe to retry the entire handler because 429 is sent from server middleware
+ * before the request is processed — no server state has changed.
+ */
+// biome-ignore lint/suspicious/noConfusingVoidType: matches Queue handler return type
+export async function withThrottleRetry(
+  fn: () => Promise<void | { timeoutSeconds: number }>
+): Promise<void | { timeoutSeconds: number }> {
+  try {
+    return await fn();
+  } catch (err) {
+    if (WorkflowAPIError.is(err) && err.status === 429) {
+      const retryAfterSeconds = Math.max(
+        // If we don't have a retry-after value, 30s seems a reasonable default
+        // to avoid re-trying during the unknown rate-limiting period.
+        1,
+        typeof err.retryAfter === 'number' ? err.retryAfter : 30
+      );
+
+      if (retryAfterSeconds < 10) {
+        runtimeLogger.warn(
+          'Throttled by workflow-server (429), retrying in-process',
+          {
+            retryAfterSeconds,
+            url: err.url,
+          }
+        );
+        // Short wait: sleep in-process, then retry once
+        await new Promise((resolve) =>
+          setTimeout(resolve, retryAfterSeconds * 1000)
+        );
+        try {
+          return await fn();
+        } catch (retryErr) {
+          // If the retry also gets throttled, defer to queue
+          if (WorkflowAPIError.is(retryErr) && retryErr.status === 429) {
+            const retryRetryAfter = Math.max(
+              1,
+              typeof retryErr.retryAfter === 'number' ? retryErr.retryAfter : 1
+            );
+            runtimeLogger.warn('Throttled again on retry, deferring to queue', {
+              retryAfterSeconds: retryRetryAfter,
+            });
+            return { timeoutSeconds: retryRetryAfter };
+          }
+          throw retryErr;
+        }
+      }
+
+      // Long wait: defer to queue infrastructure
+      runtimeLogger.warn(
+        'Throttled by workflow-server (429), deferring to queue',
+        {
+          retryAfterSeconds,
+          url: err.url,
+        }
+      );
+      return { timeoutSeconds: retryAfterSeconds };
+    }
+    throw err;
+  }
+}

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

@@ -126,6 +126,19 @@ const stepHandler = getWorldHandlers().createQueueHandler(
             step = startResult.step;
           } catch (err) {
             if (WorkflowAPIError.is(err)) {
+              if (WorkflowAPIError.is(err) && err.status === 429) {
+                const retryRetryAfter = Math.max(
+                  1,
+                  typeof err.retryAfter === 'number' ? err.retryAfter : 1
+                );
+                runtimeLogger.warn(
+                  'Throttled again on retry, deferring to queue',
+                  {
+                    retryAfterSeconds: retryRetryAfter,
+                  }
+                );
+                return { timeoutSeconds: retryRetryAfter };
+              }
               // 410 Gone: Workflow has already completed
               if (err.status === 410) {
                 console.warn(

packages/errors/src/index.ts

@@ -101,10 +101,18 @@ export class WorkflowAPIError extends WorkflowError {
   status?: number;
   code?: string;
   url?: string;
+  /** Retry-After value in seconds, present on 429 responses */
+  retryAfter?: number;
 
   constructor(
     message: string,
-    options?: { status?: number; url?: string; code?: string; cause?: unknown }
+    options?: {
+      status?: number;
+      url?: string;
+      code?: string;
+      retryAfter?: number;
+      cause?: unknown;
+    }
   ) {
     super(message, {
       cause: options?.cause,
@@ -113,6 +121,7 @@ export class WorkflowAPIError extends WorkflowError {
     this.status = options?.status;
     this.code = options?.code;
     this.url = options?.url;
+    this.retryAfter = options?.retryAfter;
   }
 
   static is(value: unknown): value is WorkflowAPIError {

packages/world-vercel/src/queue.ts

@@ -51,7 +51,7 @@ const MAX_DELAY_SECONDS = Number(
 type QueueFunction = (
   queueName: ValidQueueName,
   payload: QueuePayload,
-  opts?: QueueOptions & { delaySeconds?: number }
+  opts?: QueueOptions
 ) => ReturnType<Queue['queue']>;
 
 export function createQueue(config?: APIConfig): Queue {
@@ -71,7 +71,7 @@ export function createQueue(config?: APIConfig): Queue {
   const queue: QueueFunction = async (
     queueName,
     payload,
-    opts?: QueueOptions & { delaySeconds?: number }
+    opts?: QueueOptions
   ) => {
     // Check if we have a deployment ID either from options or environment
     const deploymentId = opts?.deploymentId ?? process.env.VERCEL_DEPLOYMENT_ID;

packages/world-vercel/src/utils.ts

@@ -6,18 +6,18 @@ import { type StructuredError, StructuredErrorSchema } from '@workflow/world';
 import { decode, encode } from 'cbor-x';
 import type { z } from 'zod';
 import {
-  trace,
+  ErrorType,
   getSpanKind,
   HttpRequestMethod,
   HttpResponseStatusCode,
-  UrlFull,
+  PeerService,
+  RpcService,
+  RpcSystem,
   ServerAddress,
   ServerPort,
-  ErrorType,
+  trace,
+  UrlFull,
   WorldParseFormat,
-  PeerService,
-  RpcSystem,
-  RpcService,
 } from './telemetry.js';
 import { version } from './version.js';
 
@@ -292,10 +292,23 @@ export async function makeRequest<T>({
             `Failed to fetch, reproduce with:\ncurl -X ${request.method} ${stringifiedHeaders} "${url}"`
           );
         }
+
+        // Parse Retry-After header for 429 responses (value is in seconds)
+        let retryAfter: number | undefined;
+        if (response.status === 429) {
+          const retryAfterHeader = response.headers.get('Retry-After');
+          if (retryAfterHeader) {
+            const parsed = parseInt(retryAfterHeader, 10);
+            if (!Number.isNaN(parsed)) {
+              retryAfter = parsed;
+            }
+          }
+        }
+
         const error = new WorkflowAPIError(
           errorData.message ||
             `${request.method} ${endpoint} -> HTTP ${response.status}: ${response.statusText}`,
-          { url, status: response.status, code: errorData.code }
+          { url, status: response.status, code: errorData.code, retryAfter }
         );
         // Record error attributes per OTEL conventions
         span?.setAttributes({

packages/world/src/queue.ts

@@ -25,6 +25,8 @@ export const WorkflowInvokePayloadSchema = z.object({
   runId: z.string(),
   traceCarrier: TraceCarrierSchema.optional(),
   requestedAt: z.coerce.date().optional(),
+  /** Number of times this message has been re-enqueued due to server errors (5xx) */
+  serverErrorRetryCount: z.number().int().optional(),
 });
 
 export const StepInvokePayloadSchema = z.object({
@@ -60,6 +62,8 @@ export interface QueueOptions {
   deploymentId?: string;
   idempotencyKey?: string;
   headers?: Record<string, string>;
+  /** Delay message delivery by this many seconds */
+  delaySeconds?: number;
 }
 
 export interface Queue {