better dequeuing from fair queue

Author: ericallam

.gitignore

@@ -62,4 +62,5 @@ apps/**/public/build
 /packages/trigger-sdk/src/package.json
 /packages/python/src/package.json
 .claude
-.mcp.log
+.mcp.log
+.cursor/debug.log

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

@@ -960,25 +960,42 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
       return;
     }
 
-    // Process one message from each selected queue
+    // Process messages from each selected tenant
+    // For fairness, process up to available concurrency slots per tenant
     for (const { tenantId, queues } of tenantQueues) {
-      for (const queueId of queues) {
-        // Check cooloff
-        if (this.cooloffEnabled && this.#isInCooloff(queueId)) {
-          continue;
-        }
+      // Get available concurrency for this tenant
+      let availableSlots = 1; // Default to 1 for backwards compatibility
+      if (this.concurrencyManager) {
+        const [current, limit] = await Promise.all([
+          this.concurrencyManager.getCurrentConcurrency("tenant", tenantId),
+          this.concurrencyManager.getConcurrencyLimit("tenant", tenantId),
+        ]);
+        availableSlots = Math.max(1, limit - current);
+      }
 
-        const processed = await this.#processOneMessage(loopId, queueId, tenantId, shardId);
+      // Process up to availableSlots messages from this tenant's queues
+      let slotsUsed = 0;
+      queueLoop: for (const queueId of queues) {
+        while (slotsUsed < availableSlots) {
+          // Check cooloff
+          if (this.cooloffEnabled && this.#isInCooloff(queueId)) {
+            break; // Try next queue
+          }
 
-        if (processed) {
-          await this.scheduler.recordProcessed?.(tenantId, queueId);
-          this.#resetCooloff(queueId);
-        } else {
-          this.#incrementCooloff(queueId);
-        }
+          const processed = await this.#processOneMessage(loopId, queueId, tenantId, shardId);
 
-        // Only process one message per queue per iteration for fairness
-        break;
+          if (processed) {
+            await this.scheduler.recordProcessed?.(tenantId, queueId);
+            this.#resetCooloff(queueId);
+            slotsUsed++;
+          } else {
+            this.#incrementCooloff(queueId);
+            break; // Queue empty or blocked, try next queue
+          }
+        }
+        if (slotsUsed >= availableSlots) {
+          break queueLoop;
+        }
       }
     }
   }