fix worker queue depth enforcement

Author: ericallam

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

@@ -1119,6 +1119,9 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
       if (depth >= this.workerQueueMaxDepth) {
         return 0;
       }
+      // Cap claim size to remaining capacity so we don't overshoot the depth limit
+      const remainingCapacity = this.workerQueueMaxDepth - depth;
+      maxClaimCount = Math.min(maxClaimCount, remainingCapacity);
     }
 
     // Claim batch of messages with visibility timeout

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

@@ -1333,7 +1333,19 @@ describe("FairQueue", () => {
           });
         }
 
-        // Start processing
+        // Start processing and track peak worker queue depth
+        let peakDepth = 0;
+        let polling = true;
+        const depthPoller = (async () => {
+          while (polling) {
+            const depth = await workerQueueManager.getLength(TEST_WORKER_QUEUE_ID);
+            if (depth > peakDepth) {
+              peakDepth = depth;
+            }
+            await new Promise((resolve) => setTimeout(resolve, 25));
+          }
+        })();
+
         queue.start();
 
         // Verify all messages eventually get processed (depth cap doesn't permanently block)
@@ -1344,6 +1356,12 @@ describe("FairQueue", () => {
           { timeout: 25000 }
         );
 
+        polling = false;
+        await depthPoller;
+
+        // Verify the depth cap was respected during processing
+        expect(peakDepth).toBeLessThanOrEqual(maxDepth);
+
         // Verify the worker queue is drained
         const finalDepth = await workerQueueManager.getLength(TEST_WORKER_QUEUE_ID);
         expect(finalDepth).toBe(0);