release reclaimed messages in batches using redis pipelines

Author: ericallam

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

@@ -106,6 +106,30 @@ export class ConcurrencyManager {
     await pipeline.exec();
   }
 
+  /**
+   * Release concurrency slots for multiple messages in a single pipeline.
+   * More efficient than calling release() multiple times.
+   */
+  async releaseBatch(
+    messages: Array<{ queue: QueueDescriptor; messageId: string }>
+  ): Promise<void> {
+    if (messages.length === 0) {
+      return;
+    }
+
+    const pipeline = this.redis.pipeline();
+
+    for (const { queue, messageId } of messages) {
+      for (const group of this.groups) {
+        const groupId = group.extractGroupId(queue);
+        const key = this.keys.concurrencyKey(group.name, groupId);
+        pipeline.srem(key, messageId);
+      }
+    }
+
+    await pipeline.exec();
+  }
+
   /**
    * Get current concurrency for a specific group.
    */

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

@@ -1349,27 +1349,26 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
         masterQueueKey: this.keys.masterQueueKey(this.masterQueue.getShardForQueue(queueId)),
       }));
 
-      // Release concurrency for each reclaimed message
+      // Release concurrency for all reclaimed messages in a single batch
       // This is critical: when a message times out, its concurrency slot must be freed
       // so the message can be processed again when it's re-claimed from the queue
       if (this.concurrencyManager && reclaimedMessages.length > 0) {
-        for (const msg of reclaimedMessages) {
-          try {
-            await this.concurrencyManager.release(
-              {
+        try {
+          await this.concurrencyManager.releaseBatch(
+            reclaimedMessages.map((msg) => ({
+              queue: {
                 id: msg.queueId,
                 tenantId: msg.tenantId,
                 metadata: msg.metadata ?? {},
               },
-              msg.messageId
-            );
-          } catch (error) {
-            this.logger.error("Failed to release concurrency for reclaimed message", {
               messageId: msg.messageId,
-              queueId: msg.queueId,
-              error: error instanceof Error ? error.message : String(error),
-            });
-          }
+            }))
+          );
+        } catch (error) {
+          this.logger.error("Failed to release concurrency for reclaimed messages", {
+            count: reclaimedMessages.length,
+            error: error instanceof Error ? error.message : String(error),
+          });
         }
       }