fix(redis-worker): close the remaining concurrency slot leaks in fair queue

matt-aitken · matt-aitken · commit b2ebf18abc31 · 2026-08-11T12:41:21.000+01:00
Four paths could strand a slot with nothing left to reclaim it.

Reclaim released the slot only after requeuing, and only for messages that
survived the requeue, so a message whose requeue threw kept its slot forever.
It now releases before the message becomes claimable, which also stops a
redelivery having its fresh reservation deleted by the previous holder.

The dead-letter branch completed the message before releasing, so anything
throwing in between stranded the slot. Release now happens first.

failMessage returned early when the stored message was missing or unparseable
without releasing, completing or requeuing, leaving the slot held.

The requeue script returned without removing the in-flight entry when the
payload was gone, so the member was rescanned on every reclaim tick forever.
Enough of them fill the scan window and starve a whole shard's reclaim.

reclaimTimedOut takes an optional pre-requeue hook; no existing caller changes.
diff --git a/packages/redis-worker/src/fair-queue/index.ts b/packages/redis-worker/src/fair-queue/index.ts
@@ -25,6 +25,7 @@ import type {
   FairScheduler,
   QueueCooloffState,
   QueueDescriptor,
+  ReclaimedMessageInfo,
   SchedulerContext,
   StoredMessage,
   TenantQueues,
@@ -1331,6 +1332,26 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
     });
   }
 
+  /**
+   * Release a concurrency slot for a message we can no longer describe, deriving the
+   * group from the queue id alone. Used on paths that bail out before the stored
+   * message is available, where the slot would otherwise be held with nothing left
+   * to reclaim it.
+   */
+  async #releaseOrphanedConcurrency(messageId: string, queueId: string): Promise<void> {
+    if (!this.concurrencyManager) {
+      return;
+    }
+
+    const descriptor: QueueDescriptor = this.queueDescriptorCache.get(queueId) ?? {
+      id: queueId,
+      tenantId: this.keys.extractTenantId(queueId),
+      metadata: {},
+    };
+
+    await this.concurrencyManager.release(descriptor, messageId);
+  }
+
   /**
    * Mark a message as failed. This will trigger retry logic if configured,
    * or move the message to the dead letter queue.
@@ -1349,6 +1370,7 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
     const dataJson = await this.redis.hget(inflightDataKey, messageId);
     if (!dataJson) {
       this.logger.error("Cannot fail message: not found in in-flight data", { messageId, queueId });
+      await this.#releaseOrphanedConcurrency(messageId, queueId);
       return;
     }
 
@@ -1360,6 +1382,7 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
         messageId,
         queueId,
       });
+      await this.#releaseOrphanedConcurrency(messageId, queueId);
       return;
     }
 
@@ -1441,13 +1464,13 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
       }
     }
 
-    // Move to DLQ
-    await this.#moveToDeadLetterQueue(storedMessage, error?.message);
-
     // Release concurrency
     if (this.concurrencyManager) {
       await this.concurrencyManager.release(descriptor, storedMessage.id);
     }
+
+    // Move to DLQ
+    await this.#moveToDeadLetterQueue(storedMessage, error?.message);
   }
 
   async #moveToDeadLetterQueue(
@@ -1521,49 +1544,54 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
     }
   }
 
-  async #reclaimTimedOutMessages(): Promise<void> {
-    let totalReclaimed = 0;
+  /**
+   * Free a reclaimed message's concurrency slot. Runs before the message is put back
+   * on the queue, so another consumer cannot reserve the same messageId and have its
+   * fresh reservation deleted by this release. Failures are logged rather than thrown
+   * so one bad message cannot stop the rest of the shard being reclaimed.
+   */
+  async #releaseReclaimedConcurrency(message: ReclaimedMessageInfo): Promise<void> {
+    if (!this.concurrencyManager) {
+      return;
+    }
 
-    for (let shardId = 0; shardId < this.shardCount; shardId++) {
-      const reclaimedMessages = await this.visibilityManager.reclaimTimedOut(shardId, (queueId) => {
-        const tenantId = this.keys.extractTenantId(queueId);
-        const dispatchShardId = this.tenantDispatch.getShardForTenant(tenantId);
-        return {
-          queueKey: this.keys.queueKey(queueId),
-          queueItemsKey: this.keys.queueItemsKey(queueId),
-          tenantQueueIndexKey: this.keys.tenantQueueIndexKey(tenantId),
-          dispatchKey: this.keys.dispatchKey(dispatchShardId),
-          tenantId,
-        };
+    try {
+      await this.concurrencyManager.release(
+        {
+          id: message.queueId,
+          tenantId: message.tenantId,
+          metadata: message.metadata ?? {},
+        },
+        message.messageId
+      );
+    } catch (error) {
+      this.logger.error("Failed to release concurrency for reclaimed message", {
+        messageId: message.messageId,
+        queueId: message.queueId,
+        error: error instanceof Error ? error.message : String(error),
       });
+    }
+  }
 
-      if (reclaimedMessages.length > 0) {
-        // 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) {
-          try {
-            await this.concurrencyManager.releaseBatch(
-              reclaimedMessages.map((msg) => ({
-                queue: {
-                  id: msg.queueId,
-                  tenantId: msg.tenantId,
-                  metadata: msg.metadata ?? {},
-                },
-                messageId: msg.messageId,
-              }))
-            );
-          } catch (error) {
-            this.logger.error("Failed to release concurrency for reclaimed messages", {
-              count: reclaimedMessages.length,
-              error: error instanceof Error ? error.message : String(error),
-            });
-          }
-        }
+  async #reclaimTimedOutMessages(): Promise<void> {
+    let totalReclaimed = 0;
 
-        // Dispatch indexes are updated atomically by the releaseMessage Lua script
-        // inside reclaimTimedOut, so no separate index update needed here.
-      }
+    for (let shardId = 0; shardId < this.shardCount; shardId++) {
+      const reclaimedMessages = await this.visibilityManager.reclaimTimedOut(
+        shardId,
+        (queueId) => {
+          const tenantId = this.keys.extractTenantId(queueId);
+          const dispatchShardId = this.tenantDispatch.getShardForTenant(tenantId);
+          return {
+            queueKey: this.keys.queueKey(queueId),
+            queueItemsKey: this.keys.queueItemsKey(queueId),
+            tenantQueueIndexKey: this.keys.tenantQueueIndexKey(tenantId),
+            dispatchKey: this.keys.dispatchKey(dispatchShardId),
+            tenantId,
+          };
+        },
+        this.#releaseReclaimedConcurrency.bind(this)
+      );
 
       totalReclaimed += reclaimedMessages.length;
     }
diff --git a/packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts b/packages/redis-worker/src/fair-queue/tests/fairQueue.test.ts
@@ -1516,5 +1516,146 @@ describe("FairQueue", () => {
         }
       }
     );
+
+    redisTest(
+      "should release the concurrency slot even when the reclaim requeue fails",
+      { timeout: 15000 },
+      async ({ redisOptions }) => {
+        keys = new DefaultFairQueueKeyProducer({ prefix: "test" });
+
+        const scheduler = new DRRScheduler({
+          redis: redisOptions,
+          keys,
+          quantum: 10,
+          maxDeficit: 100,
+        });
+
+        const queue = new TestFairQueueHelper(redisOptions, keys, {
+          scheduler,
+          payloadSchema: TestPayloadSchema,
+          shardCount: 1,
+          consumerCount: 1,
+          consumerIntervalMs: 20,
+          visibilityTimeoutMs: 200,
+          reclaimIntervalMs: 100,
+          concurrencyGroups: [
+            {
+              name: "tenant",
+              extractGroupId: (q) => q.tenantId,
+              getLimit: async () => 1,
+              defaultLimit: 1,
+            },
+          ],
+          startConsumers: false,
+        });
+
+        const redis = createRedisClient(redisOptions);
+        const queueId = "tenant:t1:queue:reclaim-fail";
+
+        try {
+          queue.onMessage(async () => {
+            await new Promise((resolve) => setTimeout(resolve, 10000));
+          });
+
+          await queue.enqueue({
+            queueId,
+            tenantId: "t1",
+            payload: { value: "msg-0" },
+          });
+
+          queue.start();
+
+          await vi.waitFor(
+            async () => {
+              expect(await redis.scard(keys.concurrencyKey("tenant", "t1"))).toBe(1);
+            },
+            { timeout: 5000 }
+          );
+
+          await redis.del(keys.queueKey(queueId));
+          await redis.set(keys.queueKey(queueId), "not-a-zset");
+
+          await vi.waitFor(
+            async () => {
+              expect(await redis.scard(keys.concurrencyKey("tenant", "t1"))).toBe(0);
+            },
+            { timeout: 8000 }
+          );
+        } finally {
+          await redis.del(keys.queueKey(queueId));
+          await redis.quit();
+          await queue.close();
+        }
+      }
+    );
+
+    redisTest(
+      "should release the concurrency slot when failMessage cannot read the message",
+      { timeout: 15000 },
+      async ({ redisOptions }) => {
+        const started: string[] = [];
+        keys = new DefaultFairQueueKeyProducer({ prefix: "test" });
+
+        const scheduler = new DRRScheduler({
+          redis: redisOptions,
+          keys,
+          quantum: 10,
+          maxDeficit: 100,
+        });
+
+        const queue = new TestFairQueueHelper(redisOptions, keys, {
+          scheduler,
+          payloadSchema: TestPayloadSchema,
+          shardCount: 1,
+          consumerCount: 1,
+          consumerIntervalMs: 20,
+          visibilityTimeoutMs: 60000,
+          concurrencyGroups: [
+            {
+              name: "tenant",
+              extractGroupId: (q) => q.tenantId,
+              getLimit: async () => 5,
+              defaultLimit: 5,
+            },
+          ],
+          startConsumers: false,
+        });
+
+        const redis = createRedisClient(redisOptions);
+
+        try {
+          queue.onMessage(async (ctx) => {
+            started.push(ctx.message.payload.value);
+            await redis.hdel(keys.inflightDataKey(0), ctx.message.id);
+            await ctx.fail(new Error("boom"));
+          });
+
+          await queue.enqueue({
+            queueId: "tenant:t1:queue:q1",
+            tenantId: "t1",
+            payload: { value: "msg-0" },
+          });
+
+          queue.start();
+
+          await vi.waitFor(
+            () => {
+              expect(started).toHaveLength(1);
+            },
+            { timeout: 10000 }
+          );
+
+          await vi.waitFor(
+            async () => {
+              expect(await redis.scard(keys.concurrencyKey("tenant", "t1"))).toBe(0);
+            },
+            { timeout: 5000 }
+          );
+        } finally {
+          await redis.quit();
+          await queue.close();
+        }
+      }
+    );
   });
 });
diff --git a/packages/redis-worker/src/fair-queue/tests/visibility.test.ts b/packages/redis-worker/src/fair-queue/tests/visibility.test.ts
@@ -912,5 +912,70 @@ describe("VisibilityManager", () => {
         await redis.quit();
       }
     );
+
+    redisTest(
+      "should drop a dangling in-flight entry whose payload is gone",
+      { timeout: 10000 },
+      async ({ redisOptions }) => {
+        keys = new DefaultFairQueueKeyProducer({ prefix: "test" });
+
+        const manager = new VisibilityManager({
+          redis: redisOptions,
+          keys,
+          shardCount: 1,
+          defaultTimeoutMs: 100,
+        });
+
+        const redis = createRedisClient(redisOptions);
+        const queueId = "tenant:t1:queue:dangling";
+        const queueKey = keys.queueKey(queueId);
+        const queueItemsKey = keys.queueItemsKey(queueId);
+        const dispatchKey = keys.dispatchKey(0);
+        const inflightKey = keys.inflightKey(0);
+        const inflightDataKey = keys.inflightDataKey(0);
+
+        try {
+          const messageId = "dangling-msg";
+          const storedMessage = {
+            id: messageId,
+            queueId,
+            tenantId: "t1",
+            payload: { id: 1, value: "test" },
+            timestamp: Date.now() - 1000,
+            attempt: 1,
+          };
+
+          await redis.zadd(queueKey, storedMessage.timestamp, messageId);
+          await redis.hset(queueItemsKey, messageId, JSON.stringify(storedMessage));
+
+          const claimResult = await manager.claim(
+            queueId,
+            queueKey,
+            queueItemsKey,
+            "consumer-1",
+            100
+          );
+          expect(claimResult.claimed).toBe(true);
+
+          await redis.hdel(inflightDataKey, messageId);
+          expect(await redis.zcard(inflightKey)).toBe(1);
+
+          await new Promise((resolve) => setTimeout(resolve, 150));
+
+          await manager.reclaimTimedOut(0, (qId) => ({
+            queueKey: keys.queueKey(qId),
+            queueItemsKey: keys.queueItemsKey(qId),
+            tenantQueueIndexKey: keys.tenantQueueIndexKey(keys.extractTenantId(qId)),
+            dispatchKey,
+            tenantId: keys.extractTenantId(qId),
+          }));
+
+          expect(await redis.zcard(inflightKey)).toBe(0);
+        } finally {
+          await manager.close();
+          await redis.quit();
+        }
+      }
+    );
   });
 });
diff --git a/packages/redis-worker/src/fair-queue/visibility.ts b/packages/redis-worker/src/fair-queue/visibility.ts
