feat: enforce max queue deliveries in handlers with graceful failure

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: pranaygp

.changeset/handler-max-deliveries.md

@@ -0,0 +1,8 @@
+---
+"@workflow/errors": patch
+"@workflow/core": patch
+"@workflow/world-local": patch
+"@workflow/builders": patch
+---
+
+Remove VQS maxDeliveries cap and enforce max delivery limit in workflow/step handlers with graceful failure

packages/builders/src/constants.ts

@@ -6,7 +6,6 @@ export const STEP_QUEUE_TRIGGER = {
   type: 'queue/v2beta' as const,
   topic: '__wkf_step_*',
   consumer: 'default',
-  maxDeliveries: 64, // Maximum number of delivery attempts (default: 3)
   retryAfterSeconds: 5, // Delay between retries (default: 60)
   initialDelaySeconds: 0, // Initial delay before first delivery (default: 0)
 };
@@ -19,7 +18,6 @@ export const WORKFLOW_QUEUE_TRIGGER = {
   type: 'queue/v2beta' as const,
   topic: '__wkf_workflow_*',
   consumer: 'default',
-  maxDeliveries: 64, // Maximum number of delivery attempts (default: 3)
   retryAfterSeconds: 5, // Delay between retries (default: 60)
   initialDelaySeconds: 0, // Initial delay before first delivery (default: 0)
 };

packages/core/src/runtime.ts

@@ -1,9 +1,11 @@
 import {
   EntityConflictError,
+  RUN_ERROR_CODES,
   RunExpiredError,
   WorkflowRuntimeError,
 } from '@workflow/errors';
 import { classifyRunError } from './classify-error.js';
+import { MAX_QUEUE_DELIVERIES } from './runtime/constants.js';
 import { parseWorkflowName } from '@workflow/utils/parse-name';
 import {
   type Event,
@@ -106,6 +108,47 @@ export function workflowEntrypoint(
       const { requestId } = metadata;
       // Extract the workflow name from the topic name
       const workflowName = metadata.queueName.slice('__wkf_workflow_'.length);
+
+      // --- Max delivery check ---
+      // Enforce max delivery limit before any infrastructure calls.
+      // This prevents runaway workflows from consuming infinite queue deliveries.
+      if (metadata.attempt > MAX_QUEUE_DELIVERIES) {
+        runtimeLogger.error(
+          `Workflow handler exceeded max deliveries (${metadata.attempt}/${MAX_QUEUE_DELIVERIES})`,
+          { workflowRunId: runId, workflowName, attempt: metadata.attempt }
+        );
+        try {
+          const world = getWorld();
+          await world.events.create(runId, {
+            eventType: 'run_failed',
+            specVersion: SPEC_VERSION_CURRENT,
+            eventData: {
+              error: {
+                message: `Workflow exceeded maximum queue deliveries (${metadata.attempt}/${MAX_QUEUE_DELIVERIES})`,
+              },
+              errorCode: RUN_ERROR_CODES.MAX_DELIVERIES_EXCEEDED,
+            },
+          });
+        } catch (err) {
+          if (
+            EntityConflictError.is(err) ||
+            RunExpiredError.is(err)
+          ) {
+            // Run already finished, consume the message
+            return;
+          }
+          runtimeLogger.error(
+            'Failed to post run_failed for max deliveries exceeded, consuming message anyway',
+            {
+              workflowRunId: runId,
+              error: err instanceof Error ? err.message : String(err),
+              attempt: metadata.attempt,
+            }
+          );
+        }
+        return;
+      }
+
       const spanLinks = await linkToCurrentContext();
 
       // Invoke user workflow within the propagated trace context and baggage

packages/core/src/runtime/constants.ts

@@ -0,0 +1 @@
+export const MAX_QUEUE_DELIVERIES = 64;

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

@@ -497,3 +497,77 @@ describe('step-handler 409 handling', () => {
     });
   });
 });
+
+describe('step-handler max deliveries', () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    vi.mocked(getStepFunction).mockReturnValue(mockStepFn);
+    mockStepFn.mockReset().mockResolvedValue('step-result');
+    mockStepFn.maxRetries = 3;
+    mockQueueMessage.mockResolvedValue(undefined);
+    vi.mocked(getWorld).mockReturnValue({
+      events: { create: mockEventsCreate },
+      queue: mockQueue,
+      getEncryptionKeyForRun: vi.fn().mockResolvedValue(undefined),
+    } as any);
+    mockEventsCreate.mockReset().mockResolvedValue({
+      step: {
+        stepId: 'step_abc',
+        status: 'running',
+        attempt: 1,
+        startedAt: new Date(),
+        input: [],
+      },
+      event: {},
+    });
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+  });
+
+  it('should post step_failed and re-queue workflow when delivery count exceeds max', async () => {
+    const result = await capturedHandler(
+      createMessage(),
+      { ...createMetadata('myStep'), attempt: 65 }
+    );
+
+    expect(result).toBeUndefined();
+    expect(mockEventsCreate).toHaveBeenCalledWith(
+      'wrun_test123',
+      expect.objectContaining({
+        eventType: 'step_failed',
+        correlationId: 'step_abc',
+      })
+    );
+    expect(mockQueueMessage).toHaveBeenCalled();
+    expect(mockRuntimeLogger.error).toHaveBeenCalledWith(
+      expect.stringContaining('exceeded max deliveries'),
+      expect.objectContaining({ workflowRunId: 'wrun_test123' })
+    );
+  });
+
+  it('should consume message silently when step_failed fails with EntityConflictError', async () => {
+    mockEventsCreate.mockRejectedValue(
+      new EntityConflictError('Step already completed')
+    );
+
+    const result = await capturedHandler(
+      createMessage(),
+      { ...createMetadata('myStep'), attempt: 65 }
+    );
+
+    expect(result).toBeUndefined();
+    expect(mockStepFn).not.toHaveBeenCalled();
+  });
+
+  it('should not trigger max deliveries check when under limit', async () => {
+    const result = await capturedHandler(
+      createMessage(),
+      { ...createMetadata('myStep'), attempt: 64 }
+    );
+
+    // Should proceed normally (step function executes)
+    expect(mockStepFn).toHaveBeenCalled();
+  });
+});