address claude code PR feedback

Author: ericallam

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

@@ -119,15 +119,34 @@ export class ConcurrencyManager {
    * Returns the minimum available capacity across all groups.
    */
   async getAvailableCapacity(queue: QueueDescriptor): Promise<number> {
-    let minCapacity = Infinity;
+    if (this.groups.length === 0) {
+      return 0;
+    }
 
-    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);
+    // Build group data for parallel fetching
+    const groupData = this.groups.map((group) => ({
+      group,
+      groupId: group.extractGroupId(queue),
+    }));
+
+    // Fetch all current counts and limits in parallel
+    const [currents, limits] = await Promise.all([
+      Promise.all(
+        groupData.map(({ group, groupId }) =>
+          this.redis.scard(this.keys.concurrencyKey(group.name, groupId))
+        )
+      ),
+      Promise.all(
+        groupData.map(({ group, groupId }) =>
+          group.getLimit(groupId).then((limit) => limit || group.defaultLimit)
+        )
+      ),
+    ]);
 
+    // Calculate minimum available capacity across all groups
+    let minCapacity = Infinity;
+    for (let i = 0; i < groupData.length; i++) {
+      const available = Math.max(0, limits[i]! - currents[i]!);
       minCapacity = Math.min(minCapacity, available);
     }
 

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

@@ -84,6 +84,7 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
   private workerQueueEnabled: boolean;
   private workerQueueBlockingTimeoutSeconds: number;
   private workerQueueResolver?: (message: StoredMessage<z.infer<TPayloadSchema>>) => string;
+  private batchClaimSize: number;
 
   // Cooloff state
   private cooloffEnabled: boolean;
@@ -139,6 +140,9 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
     this.workerQueueBlockingTimeoutSeconds = options.workerQueue?.blockingTimeoutSeconds ?? 10;
     this.workerQueueResolver = options.workerQueue?.resolveWorkerQueue;
 
+    // Batch claiming
+    this.batchClaimSize = options.batchClaimSize ?? 10;
+
     // Cooloff
     this.cooloffEnabled = options.cooloff?.enabled ?? true;
     this.cooloffThreshold = options.cooloff?.threshold ?? 10;
@@ -892,8 +896,20 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
           messagesProcessed += processedFromQueue;
           this.batchedSpanManager.incrementStat(loopId, "messages_claimed", processedFromQueue);
 
-          if (this.scheduler.recordProcessed) {
-            // Record each processed message for DRR deficit tracking
+          // Record processed messages for DRR deficit tracking
+          // Use batch variant if available for efficiency, otherwise fall back to single calls
+          if (this.scheduler.recordProcessedBatch) {
+            await this.telemetry.trace(
+              "recordProcessedBatch",
+              async (span) => {
+                span.setAttribute(FairQueueAttributes.QUEUE_ID, queueId);
+                span.setAttribute(FairQueueAttributes.TENANT_ID, tenantId);
+                span.setAttribute("count", processedFromQueue);
+                await this.scheduler.recordProcessedBatch!(tenantId, queueId, processedFromQueue);
+              },
+              { kind: SpanKind.INTERNAL }
+            );
+          } else if (this.scheduler.recordProcessed) {
             for (let i = 0; i < processedFromQueue; i++) {
               await this.telemetry.trace(
                 "recordProcessed",
@@ -936,7 +952,7 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
     };
 
     // Determine how many messages we can claim based on concurrency
-    let maxClaimCount = 10; // Default batch size cap
+    let maxClaimCount = this.batchClaimSize;
     if (this.concurrencyManager) {
       const availableCapacity = await this.concurrencyManager.getAvailableCapacity(descriptor);
       if (availableCapacity === 0) {
@@ -974,7 +990,12 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
       return 0;
     }
 
-    // Use single shared worker queue - all messages go to one queue, all consumers pop atomically
+    // Single shared worker queue pattern:
+    // All consumers pop from one queue ("worker-queue") for atomic distribution.
+    // Trade-off: Simpler code and fair distribution vs. potential contention
+    // under very high load (>10k messages/sec). For most workloads, Redis
+    // can handle 100k+ ops/sec on a single key, so this is rarely a bottleneck.
+    // Future: Consider adding optional worker queue sharding if needed.
     const workerQueueId = "worker-queue";
     let processedCount = 0;
 

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

@@ -28,6 +28,15 @@ export abstract class BaseScheduler implements FairScheduler {
     // Default: no state tracking
   }
 
+  /**
+   * Called after processing multiple messages to update scheduler state.
+   * Batch variant for efficiency - reduces Redis calls when processing multiple messages.
+   * Default implementation does nothing.
+   */
+  async recordProcessedBatch(_tenantId: string, _queueId: string, _count: number): Promise<void> {
+    // Default: no state tracking
+  }
+
   /**
    * Initialize the scheduler.
    * Default implementation does nothing.

packages/redis-worker/src/fair-queue/schedulers/drr.ts

@@ -140,6 +140,18 @@ export class DRRScheduler extends BaseScheduler {
     await this.#decrementDeficit(tenantId);
   }
 
+  /**
+   * Record that multiple messages were processed from a tenant.
+   * Decrements the tenant's deficit by count atomically.
+   */
+  override async recordProcessedBatch(
+    tenantId: string,
+    _queueId: string,
+    count: number
+  ): Promise<void> {
+    await this.#decrementDeficitBatch(tenantId, count);
+  }
+
   override async close(): Promise<void> {
     await this.redis.quit();
   }
@@ -249,6 +261,17 @@ export class DRRScheduler extends BaseScheduler {
     return parseFloat(result);
   }
 
+  /**
+   * Decrement deficit for a tenant by a count atomically.
+   */
+  async #decrementDeficitBatch(tenantId: string, count: number): Promise<number> {
+    const key = this.#deficitKey();
+
+    // Use Lua script to decrement by count and ensure non-negative
+    const result = await this.redis.drrDecrementDeficitBatch(key, tenantId, count.toString());
+    return parseFloat(result);
+  }
+
   #registerCommands(): void {
     // Atomic quantum addition with capping for multiple tenants
     this.redis.defineCommand("drrAddQuantum", {
@@ -295,6 +318,27 @@ if newDeficit < 0 then
   newDeficit = 0
 end
 
+return tostring(newDeficit)
+      `,
+    });
+
+    // Atomic deficit decrement by count with floor at 0
+    this.redis.defineCommand("drrDecrementDeficitBatch", {
+      numberOfKeys: 1,
+      lua: `
+local deficitKey = KEYS[1]
+local tenantId = ARGV[1]
+local count = tonumber(ARGV[2])
+
+local newDeficit = redis.call('HINCRBYFLOAT', deficitKey, tenantId, -count)
+newDeficit = tonumber(newDeficit)
+
+-- Floor at 0
+if newDeficit < 0 then
+  redis.call('HSET', deficitKey, tenantId, 0)
+  newDeficit = 0
+end
+
 return tostring(newDeficit)
       `,
     });
@@ -312,6 +356,12 @@ declare module "@internal/redis" {
     ): Promise<string[]>;
 
     drrDecrementDeficit(deficitKey: string, tenantId: string): Promise<string>;
+
+    drrDecrementDeficitBatch(
+      deficitKey: string,
+      tenantId: string,
+      count: string
+    ): Promise<string>;
   }
 }
 

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

@@ -222,6 +222,167 @@ describe("ConcurrencyManager", () => {
     );
   });
 
+  describe("getAvailableCapacity", () => {
+    redisTest(
+      "should return available capacity for single group",
+      { timeout: 10000 },
+      async ({ redisOptions }) => {
+        keys = new DefaultFairQueueKeyProducer({ prefix: "test" });
+
+        const manager = new ConcurrencyManager({
+          redis: redisOptions,
+          keys,
+          groups: [
+            {
+              name: "tenant",
+              extractGroupId: (q) => q.tenantId,
+              getLimit: async () => 10,
+              defaultLimit: 10,
+            },
+          ],
+        });
+
+        const queue: QueueDescriptor = {
+          id: "queue-1",
+          tenantId: "t1",
+          metadata: {},
+        };
+
+        // Initial capacity should be full
+        let capacity = await manager.getAvailableCapacity(queue);
+        expect(capacity).toBe(10);
+
+        // Reserve 3 slots
+        await manager.reserve(queue, "msg-1");
+        await manager.reserve(queue, "msg-2");
+        await manager.reserve(queue, "msg-3");
+
+        // Capacity should be reduced
+        capacity = await manager.getAvailableCapacity(queue);
+        expect(capacity).toBe(7);
+
+        await manager.close();
+      }
+    );
+
+    redisTest(
+      "should return minimum capacity across multiple groups",
+      { timeout: 10000 },
+      async ({ redisOptions }) => {
+        keys = new DefaultFairQueueKeyProducer({ prefix: "test" });
+
+        const manager = new ConcurrencyManager({
+          redis: redisOptions,
+          keys,
+          groups: [
+            {
+              name: "tenant",
+              extractGroupId: (q) => q.tenantId,
+              getLimit: async () => 5,
+              defaultLimit: 5,
+            },
+            {
+              name: "organization",
+              extractGroupId: (q) => (q.metadata.orgId as string) ?? "default",
+              getLimit: async () => 20,
+              defaultLimit: 20,
+            },
+          ],
+        });
+
+        const queue: QueueDescriptor = {
+          id: "queue-1",
+          tenantId: "t1",
+          metadata: { orgId: "org1" },
+        };
+
+        // Initial capacity should be minimum (5 for tenant, 20 for org)
+        let capacity = await manager.getAvailableCapacity(queue);
+        expect(capacity).toBe(5);
+
+        // Reserve 3 slots
+        await manager.reserve(queue, "msg-1");
+        await manager.reserve(queue, "msg-2");
+        await manager.reserve(queue, "msg-3");
+
+        // Now tenant has 2 left, org has 17 left - minimum is 2
+        capacity = await manager.getAvailableCapacity(queue);
+        expect(capacity).toBe(2);
+
+        await manager.close();
+      }
+    );
+
+    redisTest(
+      "should return 0 when any group is at capacity",
+      { timeout: 10000 },
+      async ({ redisOptions }) => {
+        keys = new DefaultFairQueueKeyProducer({ prefix: "test" });
+
+        const manager = new ConcurrencyManager({
+          redis: redisOptions,
+          keys,
+          groups: [
+            {
+              name: "tenant",
+              extractGroupId: (q) => q.tenantId,
+              getLimit: async () => 3,
+              defaultLimit: 3,
+            },
+            {
+              name: "organization",
+              extractGroupId: (q) => (q.metadata.orgId as string) ?? "default",
+              getLimit: async () => 10,
+              defaultLimit: 10,
+            },
+          ],
+        });
+
+        const queue: QueueDescriptor = {
+          id: "queue-1",
+          tenantId: "t1",
+          metadata: { orgId: "org1" },
+        };
+
+        // Fill up tenant capacity
+        await manager.reserve(queue, "msg-1");
+        await manager.reserve(queue, "msg-2");
+        await manager.reserve(queue, "msg-3");
+
+        // Tenant is at 3/3, org is at 3/10
+        const capacity = await manager.getAvailableCapacity(queue);
+        expect(capacity).toBe(0);
+
+        await manager.close();
+      }
+    );
+
+    redisTest(
+      "should return 0 when no groups are configured",
+      { timeout: 10000 },
+      async ({ redisOptions }) => {
+        keys = new DefaultFairQueueKeyProducer({ prefix: "test" });
+
+        const manager = new ConcurrencyManager({
+          redis: redisOptions,
+          keys,
+          groups: [],
+        });
+
+        const queue: QueueDescriptor = {
+          id: "queue-1",
+          tenantId: "t1",
+          metadata: {},
+        };
+
+        const capacity = await manager.getAvailableCapacity(queue);
+        expect(capacity).toBe(0);
+
+        await manager.close();
+      }
+    );
+  });
+
   describe("atomic reservation", () => {
     redisTest(
       "should atomically reserve across groups",