fix(batch): optimize processing batch trigger v2

Author: ericallam

apps/webapp/app/env.server.ts

@@ -951,6 +951,12 @@ const EnvironmentSchema = z
     BATCH_QUEUE_MAX_DEFICIT: z.coerce.number().int().default(100),
     BATCH_QUEUE_CONSUMER_COUNT: z.coerce.number().int().default(3),
     BATCH_QUEUE_CONSUMER_INTERVAL_MS: z.coerce.number().int().default(50),
+    // Number of master queue shards for horizontal scaling
+    BATCH_QUEUE_SHARD_COUNT: z.coerce.number().int().default(1),
+    // Maximum queues to fetch from master queue per iteration
+    BATCH_QUEUE_MASTER_QUEUE_LIMIT: z.coerce.number().int().default(1000),
+    // Worker queue blocking timeout in seconds (for two-stage processing)
+    BATCH_QUEUE_WORKER_QUEUE_TIMEOUT_SECONDS: z.coerce.number().int().default(10),
     // 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(),

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

@@ -171,7 +171,10 @@ function createRunEngine() {
       drr: {
         quantum: env.BATCH_QUEUE_DRR_QUANTUM,
         maxDeficit: env.BATCH_QUEUE_MAX_DEFICIT,
+        masterQueueLimit: env.BATCH_QUEUE_MASTER_QUEUE_LIMIT,
       },
+      shardCount: env.BATCH_QUEUE_SHARD_COUNT,
+      workerQueueBlockingTimeoutSeconds: env.BATCH_QUEUE_WORKER_QUEUE_TIMEOUT_SECONDS,
       consumerCount: env.BATCH_QUEUE_CONSUMER_COUNT,
       consumerIntervalMs: env.BATCH_QUEUE_CONSUMER_INTERVAL_MS,
       // Default processing concurrency when no specific limit is set

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

@@ -108,6 +108,7 @@ export class BatchQueue {
       keys: keyProducer,
       quantum: options.drr.quantum,
       maxDeficit: options.drr.maxDeficit,
+      masterQueueLimit: options.drr.masterQueueLimit,
       logger: {
         debug: (msg, ctx) => this.logger.debug(msg, ctx),
         error: (msg, ctx) => this.logger.error(msg, ctx),
@@ -121,7 +122,7 @@ export class BatchQueue {
       scheduler,
       payloadSchema: BatchItemPayloadSchema,
       validateOnEnqueue: false, // We control the payload
-      shardCount: 1, // Batches don't need sharding
+      shardCount: options.shardCount ?? 1,
       consumerCount: options.consumerCount,
       consumerIntervalMs: options.consumerIntervalMs,
       visibilityTimeoutMs: 60_000, // 1 minute for batch item processing
@@ -131,6 +132,14 @@ export class BatchQueue {
         threshold: 5,
         periodMs: 5_000,
       },
+      // Enable two-stage processing with worker queues for better parallelism (when configured)
+      // Worker queues provide better concurrency by separating queue selection from message processing
+      workerQueue: options.workerQueueBlockingTimeoutSeconds
+        ? {
+            enabled: true,
+            blockingTimeoutSeconds: options.workerQueueBlockingTimeoutSeconds,
+          }
+        : undefined,
       // Concurrency group based on tenant (environment)
       // This limits how many batch items can be processed concurrently per environment
       // Items wait in queue until capacity frees up

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

@@ -123,6 +123,8 @@ export type DRRConfig = {
   quantum: number;
   /** Maximum accumulated deficit (prevents starvation) */
   maxDeficit: number;
+  /** Maximum queues to fetch from master queue (default: 1000) */
+  masterQueueLimit?: number;
 };
 
 // ============================================================================
@@ -196,6 +198,10 @@ export type BatchQueueOptions = {
   consumerCount: number;
   /** Interval between consumer iterations (ms) */
   consumerIntervalMs: number;
+  /** Number of master queue shards (default: 1) */
+  shardCount?: number;
+  /** Worker queue blocking timeout in seconds (enables two-stage processing) */
+  workerQueueBlockingTimeoutSeconds?: number;
   /** Whether to start consumers on initialization */
   startConsumers?: boolean;
   /**

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

@@ -340,7 +340,10 @@ export class RunEngine {
       drr: {
         quantum: options.batchQueue?.drr?.quantum ?? 5,
         maxDeficit: options.batchQueue?.drr?.maxDeficit ?? 50,
+        masterQueueLimit: options.batchQueue?.drr?.masterQueueLimit,
       },
+      shardCount: options.batchQueue?.shardCount,
+      workerQueueBlockingTimeoutSeconds: options.batchQueue?.workerQueueBlockingTimeoutSeconds,
       consumerCount: options.batchQueue?.consumerCount ?? 2,
       consumerIntervalMs: options.batchQueue?.consumerIntervalMs ?? 100,
       defaultConcurrency: options.batchQueue?.defaultConcurrency ?? 10,

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

@@ -76,6 +76,10 @@ export type RunEngineOptions = {
   batchQueue?: {
     redis: RedisOptions;
     drr?: Partial<DRRConfig>;
+    /** Number of master queue shards (default: 1) */
+    shardCount?: number;
+    /** Worker queue blocking timeout in seconds (enables two-stage processing) */
+    workerQueueBlockingTimeoutSeconds?: number;
     consumerCount?: number;
     consumerIntervalMs?: number;
     /** Default processing concurrency per environment when no specific limit is set */

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

@@ -114,6 +114,26 @@ export class ConcurrencyManager {
     return await this.redis.scard(key);
   }
 
+  /**
+   * Get available capacity for a queue across all concurrency groups.
+   * Returns the minimum available capacity across all groups.
+   */
+  async getAvailableCapacity(queue: QueueDescriptor): Promise<number> {
+    let minCapacity = Infinity;
+
+    for (const group of this.groups) {
+      const groupId = group.extractGroupId(queue);
+      const key = this.keys.concurrencyKey(group.name, groupId);
+      const current = await this.redis.scard(key);
+      const limit = (await group.getLimit(groupId)) || group.defaultLimit;
+      const available = Math.max(0, limit - current);
+
+      minCapacity = Math.min(minCapacity, available);
+    }
+
+    return minCapacity === Infinity ? 0 : minCapacity;
+  }
+
   /**
    * Get concurrency limit for a specific group.
    */