fix(batch): move batch queue global rate limiter to worker consumer level

The global rate limiter was being applied at the FairQueue claim phase,
  consuming 1 token per queue-claim-attempt rather than per item processed.
  With many small queues (each batch is its own queue), consumers burned
  through tokens on empty or single-item queues, causing aggressive
  throttling well below the intended items/sec limit.

  Changes:
  - Move rate limiter from FairQueue claim phase to BatchQueue worker queue
    consumer loop (before blockingPop), so each token = 1 item processed
  - Replace the FairQueue rate limiter with a worker queue depth cap to
    prevent unbounded growth that could cause visibility timeouts
  - Add BATCH_QUEUE_WORKER_QUEUE_MAX_DEPTH env var (optional, disabled by default)

Author: ericallam

.server-changes/batch-rate-limiter-fix.md

@@ -0,0 +1,6 @@
+---
+area: webapp
+type: fix
+---
+
+Move batch queue global rate limiter from FairQueue claim phase to BatchQueue worker queue consumer for accurate per-item rate limiting. Add worker queue depth cap to prevent unbounded growth that could cause visibility timeouts.

apps/webapp/app/env.server.ts

@@ -993,6 +993,9 @@ const EnvironmentSchema = z
     // Global rate limit: max items processed per second across all consumers
     // If not set, no global rate limiting is applied
     BATCH_QUEUE_GLOBAL_RATE_LIMIT: z.coerce.number().int().positive().optional(),
+    // Max items in the worker queue before claiming pauses (protects visibility timeouts)
+    // If not set, no depth limit is applied
+    BATCH_QUEUE_WORKER_QUEUE_MAX_DEPTH: z.coerce.number().int().positive().optional(),
 
     ADMIN_WORKER_ENABLED: z.string().default(process.env.WORKER_ENABLED ?? "true"),
     ADMIN_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(2),

apps/webapp/app/v3/runEngine.server.ts

@@ -198,6 +198,8 @@ function createRunEngine() {
       globalRateLimiter: env.BATCH_QUEUE_GLOBAL_RATE_LIMIT
         ? createBatchGlobalRateLimiter(env.BATCH_QUEUE_GLOBAL_RATE_LIMIT)
         : undefined,
+      // Worker queue depth cap - prevents unbounded growth protecting visibility timeouts
+      workerQueueMaxDepth: env.BATCH_QUEUE_WORKER_QUEUE_MAX_DEPTH,
       retry: {
         maxAttempts: 6,
         minTimeoutInMs: 1_000,

internal-packages/run-engine/src/batch-queue/index.ts

@@ -18,6 +18,7 @@ import {
   isAbortError,
   WorkerQueueManager,
   type FairQueueOptions,
+  type GlobalRateLimiter,
 } from "@trigger.dev/redis-worker";
 import { BatchCompletionTracker } from "./completionTracker.js";
 import type {
@@ -76,6 +77,7 @@ export class BatchQueue {
   private abortController: AbortController;
   private workerQueueConsumerLoops: Promise<void>[] = [];
   private workerQueueBlockingTimeoutSeconds: number;
+  private globalRateLimiter?: GlobalRateLimiter;
   private batchedSpanManager: BatchedSpanManager;
 
   // Metrics
@@ -95,6 +97,7 @@ export class BatchQueue {
     this.maxAttempts = options.retry?.maxAttempts ?? 1;
     this.abortController = new AbortController();
     this.workerQueueBlockingTimeoutSeconds = options.workerQueueBlockingTimeoutSeconds ?? 10;
+    this.globalRateLimiter = options.globalRateLimiter;
 
     // Initialize metrics if meter is provided
     if (options.meter) {
@@ -174,8 +177,9 @@ export class BatchQueue {
           },
         },
       ],
-      // Optional global rate limiter to limit max items/sec across all consumers
-      globalRateLimiter: options.globalRateLimiter,
+      // Worker queue depth cap to prevent unbounded growth (protects visibility timeouts)
+      workerQueueMaxDepth: options.workerQueueMaxDepth,
+      workerQueueDepthCheckId: BATCH_WORKER_QUEUE_ID,
       // Enable retry with DLQ disabled when retry config is provided.
       // BatchQueue handles the "final failure" in its own processing loop,
       // so we don't need the DLQ - we just need the retry scheduling.
@@ -641,6 +645,17 @@ export class BatchQueue {
         }
 
         try {
+          // Rate limit per-item at the processing level (1 token per message)
+          if (this.globalRateLimiter) {
+            const result = await this.globalRateLimiter.limit();
+            if (!result.allowed && result.resetAt) {
+              const waitMs = Math.max(0, result.resetAt - Date.now());
+              if (waitMs > 0) {
+                await new Promise((resolve) => setTimeout(resolve, waitMs));
+              }
+            }
+          }
+
           await this.batchedSpanManager.withBatchedSpan(
             loopId,
             async (span) => {

internal-packages/run-engine/src/batch-queue/tests/index.test.ts

@@ -1,6 +1,7 @@
 import { redisTest } from "@internal/testcontainers";
 import { describe, expect, vi } from "vitest";
 import { BatchQueue } from "../index.js";
+import type { GlobalRateLimiter } from "@trigger.dev/redis-worker";
 import type { CompleteBatchResult, InitializeBatchOptions, BatchItem } from "../types.js";
 
 vi.setConfig({ testTimeout: 60_000 });
@@ -658,4 +659,121 @@ describe("BatchQueue", () => {
       }
     );
   });
+
+  describe("global rate limiter at worker queue consumer level", () => {
+    redisTest(
+      "should call rate limiter once per item processed",
+      async ({ redisContainer }) => {
+        let limitCallCount = 0;
+        const rateLimiter: GlobalRateLimiter = {
+          async limit() {
+            limitCallCount++;
+            return { allowed: true };
+          },
+        };
+
+        const queue = new BatchQueue({
+          redis: {
+            host: redisContainer.getHost(),
+            port: redisContainer.getPort(),
+            keyPrefix: "test:",
+          },
+          drr: { quantum: 5, maxDeficit: 50 },
+          consumerCount: 1,
+          consumerIntervalMs: 50,
+          startConsumers: true,
+          globalRateLimiter: rateLimiter,
+        });
+
+        let completionResult: CompleteBatchResult | null = null;
+
+        try {
+          queue.onProcessItem(async ({ itemIndex }) => {
+            return { success: true, runId: `run_${itemIndex}` };
+          });
+
+          queue.onBatchComplete(async (result) => {
+            completionResult = result;
+          });
+
+          const itemCount = 5;
+          await queue.initializeBatch(createInitOptions("batch1", "env1", itemCount));
+          await enqueueItems(queue, "batch1", "env1", createBatchItems(itemCount));
+
+          await vi.waitFor(
+            () => {
+              expect(completionResult).not.toBeNull();
+            },
+            { timeout: 10000 }
+          );
+
+          expect(completionResult!.successfulRunCount).toBe(itemCount);
+          // Rate limiter should be called at least once per item processed
+          // (may be called more due to consumer loop iterations with empty pops)
+          expect(limitCallCount).toBeGreaterThanOrEqual(itemCount);
+        } finally {
+          await queue.close();
+        }
+      }
+    );
+
+    redisTest(
+      "should delay processing when rate limited",
+      async ({ redisContainer }) => {
+        let limitCallCount = 0;
+        const rateLimiter: GlobalRateLimiter = {
+          async limit() {
+            limitCallCount++;
+            // Rate limit the first 3 calls, then allow
+            if (limitCallCount <= 3) {
+              return { allowed: false, resetAt: Date.now() + 100 };
+            }
+            return { allowed: true };
+          },
+        };
+
+        const queue = new BatchQueue({
+          redis: {
+            host: redisContainer.getHost(),
+            port: redisContainer.getPort(),
+            keyPrefix: "test:",
+          },
+          drr: { quantum: 5, maxDeficit: 50 },
+          consumerCount: 1,
+          consumerIntervalMs: 50,
+          startConsumers: true,
+          globalRateLimiter: rateLimiter,
+        });
+
+        let completionResult: CompleteBatchResult | null = null;
+
+        try {
+          queue.onProcessItem(async ({ itemIndex }) => {
+            return { success: true, runId: `run_${itemIndex}` };
+          });
+
+          queue.onBatchComplete(async (result) => {
+            completionResult = result;
+          });
+
+          await queue.initializeBatch(createInitOptions("batch1", "env1", 3));
+          await enqueueItems(queue, "batch1", "env1", createBatchItems(3));
+
+          // Should still complete despite initial rate limiting
+          await vi.waitFor(
+            () => {
+              expect(completionResult).not.toBeNull();
+            },
+            { timeout: 10000 }
+          );
+
+          expect(completionResult!.successfulRunCount).toBe(3);
+          // Rate limiter was called more times than items due to initial rejections
+          expect(limitCallCount).toBeGreaterThan(3);
+        } finally {
+          await queue.close();
+        }
+      }
+    );
+  });
 });

internal-packages/run-engine/src/batch-queue/types.ts

@@ -213,8 +213,15 @@ export type BatchQueueOptions = {
   /**
    * Optional global rate limiter to limit processing across all consumers.
    * When configured, limits the max items/second processed globally.
+   * Rate limiting happens at the worker queue consumer level (1 token per item).
    */
   globalRateLimiter?: GlobalRateLimiter;
+  /**
+   * Maximum number of items allowed in the worker queue before claiming pauses.
+   * Prevents unbounded worker queue growth which could cause visibility timeouts.
+   * Disabled by default (undefined = no limit).
+   */
+  workerQueueMaxDepth?: number;
   /** Logger instance */
   logger?: Logger;
   logLevel?: LogLevel;

internal-packages/run-engine/src/engine/index.ts

@@ -398,6 +398,7 @@ export class RunEngine {
       consumerIntervalMs: options.batchQueue?.consumerIntervalMs ?? 100,
       defaultConcurrency: options.batchQueue?.defaultConcurrency ?? 10,
       globalRateLimiter: options.batchQueue?.globalRateLimiter,
+      workerQueueMaxDepth: options.batchQueue?.workerQueueMaxDepth,
       startConsumers: startBatchQueueConsumers,
       retry: options.batchQueue?.retry,
       tracer: options.tracer,

internal-packages/run-engine/src/engine/types.ts

@@ -109,6 +109,8 @@ export type RunEngineOptions = {
     defaultConcurrency?: number;
     /** Optional global rate limiter to limit processing across all consumers */
     globalRateLimiter?: GlobalRateLimiter;
+    /** Maximum worker queue depth before claiming pauses (protects visibility timeouts) */
+    workerQueueMaxDepth?: number;
     /** Retry configuration for failed batch items */
     retry?: {
       /** Maximum number of attempts (including the first). Default: 1 (no retries) */

packages/redis-worker/src/fair-queue/index.ts

@@ -24,7 +24,6 @@ import type {
   FairQueueKeyProducer,
   FairQueueOptions,
   FairScheduler,
-  GlobalRateLimiter,
   QueueCooloffState,
   QueueDescriptor,
   SchedulerContext,
@@ -97,8 +96,9 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
   private maxCooloffStatesSize: number;
   private queueCooloffStates = new Map<string, QueueCooloffState>();
 
-  // Global rate limiter
-  private globalRateLimiter?: GlobalRateLimiter;
+  // Worker queue backpressure
+  private workerQueueMaxDepth: number;
+  private workerQueueDepthCheckId?: string;
 
   // Consumer tracing
   private consumerTraceMaxIterations: number;
@@ -152,8 +152,9 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
     this.cooloffPeriodMs = options.cooloff?.periodMs ?? 10_000;
     this.maxCooloffStatesSize = options.cooloff?.maxStatesSize ?? 1000;
 
-    // Global rate limiter
-    this.globalRateLimiter = options.globalRateLimiter;
+    // Worker queue backpressure
+    this.workerQueueMaxDepth = options.workerQueueMaxDepth ?? 0;
+    this.workerQueueDepthCheckId = options.workerQueueDepthCheckId;
 
     // Consumer tracing
     this.consumerTraceMaxIterations = options.consumerTraceMaxIterations ?? 500;
@@ -1110,15 +1111,13 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
       maxClaimCount = Math.min(maxClaimCount, availableCapacity);
     }
 
-    // Check global rate limit - wait if rate limited
-    if (this.globalRateLimiter) {
-      const result = await this.globalRateLimiter.limit();
-      if (!result.allowed && result.resetAt) {
-        const waitMs = Math.max(0, result.resetAt - Date.now());
-        if (waitMs > 0) {
-          this.logger.debug("Global rate limit reached, waiting", { waitMs, loopId });
-          await new Promise((resolve) => setTimeout(resolve, waitMs));
-        }
+    // Check worker queue depth to prevent unbounded growth.
+    // Messages in the worker queue are already in-flight with a visibility timeout.
+    // If the queue is too deep, consumers can't keep up, and messages risk timing out.
+    if (this.workerQueueMaxDepth > 0 && this.workerQueueDepthCheckId) {
+      const depth = await this.workerQueueManager.getLength(this.workerQueueDepthCheckId);
+      if (depth >= this.workerQueueMaxDepth) {
+        return 0;
       }
     }
 

packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts

@@ -1282,4 +1282,76 @@ describe("FairQueue", () => {
     );
   });
 
+  describe("worker queue depth cap", () => {
+    redisTest(
+      "should respect worker queue max depth and resume after draining",
+      { timeout: 30000 },
+      async ({ redisOptions }) => {
+        const processed: string[] = [];
+        keys = new DefaultFairQueueKeyProducer({ prefix: "test" });
+
+        const scheduler = new DRRScheduler({
+          redis: redisOptions,
+          keys,
+          quantum: 100,
+          maxDeficit: 1000,
+        });
+
+        const workerQueueManager = new WorkerQueueManager({
+          redis: redisOptions,
+          keys,
+        });
+
+        // Create FairQueue with a small depth cap
+        const maxDepth = 3;
+        const queue = new TestFairQueueHelper(redisOptions, keys, {
+          scheduler,
+          payloadSchema: TestPayloadSchema,
+          shardCount: 1,
+          consumerCount: 1,
+          consumerIntervalMs: 50,
+          visibilityTimeoutMs: 30000,
+          workerQueueMaxDepth: maxDepth,
+          workerQueueDepthCheckId: TEST_WORKER_QUEUE_ID,
+          startConsumers: false,
+        });
+
+        // Use a slow handler to let the worker queue build up
+        queue.onMessage(async (ctx) => {
+          await new Promise((resolve) => setTimeout(resolve, 200));
+          processed.push(ctx.message.payload.value);
+          await ctx.complete();
+        });
+
+        // Enqueue 10 messages
+        const totalMessages = 10;
+        for (let i = 0; i < totalMessages; i++) {
+          await queue.enqueue({
+            queueId: "tenant:t1:queue:q1",
+            tenantId: "t1",
+            payload: { value: `msg-${i}` },
+          });
+        }
+
+        // Start processing
+        queue.start();
+
+        // Verify all messages eventually get processed (depth cap doesn't permanently block)
+        await vi.waitFor(
+          () => {
+            expect(processed.length).toBe(totalMessages);
+          },
+          { timeout: 25000 }
+        );
+
+        // Verify the worker queue is drained
+        const finalDepth = await workerQueueManager.getLength(TEST_WORKER_QUEUE_ID);
+        expect(finalDepth).toBe(0);
+
+        await workerQueueManager.close();
+        await queue.close();
+      }
+    );
+  });
+
 });