fix(redis-worker): stop fair queue leaking concurrency slots

A slot was only released when the message's in-flight record could still be
read, and the release ran after that record had already been deleted. Any
failure in between left the slot held with nothing able to reclaim it. Once a
tenant leaked its whole limit, every queue it owned stalled for good.

Author: matt-aitken

.changeset/fair-queue-concurrency-slot-leak.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/redis-worker": patch
+---
+
+Fair queue consumers no longer leak concurrency slots. A slot is now always released when a message completes or is put back on the queue, even when its in-flight record has already gone. Leaked slots were never reclaimed, so enough of them would permanently stall every queue belonging to that tenant.

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

@@ -1253,14 +1253,14 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
         })
       : { id: queueId, tenantId: this.keys.extractTenantId(queueId), metadata: {} };
 
-    // Complete in visibility manager
-    await this.visibilityManager.complete(messageId, queueId);
-
     // Release concurrency
-    if (this.concurrencyManager && storedMessage) {
+    if (this.concurrencyManager) {
       await this.concurrencyManager.release(descriptor, messageId);
     }
 
+    // Complete in visibility manager
+    await this.visibilityManager.complete(messageId, queueId);
+
     // Update both old and new indexes, clean up caches if queue is empty
     const { queueEmpty } = await this.#updateAllIndexesAfterDequeue(queueId, descriptor.tenantId);
     if (queueEmpty) {
@@ -1308,6 +1308,11 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
         })
       : { id: queueId, tenantId: this.keys.extractTenantId(queueId), metadata: {} };
 
+    // Release concurrency
+    if (this.concurrencyManager) {
+      await this.concurrencyManager.release(descriptor, messageId);
+    }
+
     // Release back to queue (visibility manager updates dispatch indexes atomically)
     // Dispatch shard is tenant-based, not queue-based
     const dispatchShardId = this.tenantDispatch.getShardForTenant(descriptor.tenantId);
@@ -1324,11 +1329,6 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
       Date.now() // Put at back of queue
     );
 
-    // Release concurrency
-    if (this.concurrencyManager && storedMessage) {
-      await this.concurrencyManager.release(descriptor, messageId);
-    }
-
     this.logger.debug("Message released", {
       messageId,
       queueId,

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

@@ -10,7 +10,7 @@ import {
   WorkerQueueManager,
 } from "../index.js";
 import type { FairQueueKeyProducer, FairQueueOptions } from "../types.js";
-import type { RedisOptions } from "@internal/redis";
+import { createRedisClient, type RedisOptions } from "@internal/redis";
 
 // Define a common payload schema for tests
 const TestPayloadSchema = z.object({ value: z.string() });
@@ -1370,4 +1370,73 @@ describe("FairQueue", () => {
       }
     );
   });
+
+  describe("concurrency slot release", () => {
+    redisTest(
+      "should release the concurrency slot when the in-flight record is already gone",
+      { timeout: 15000 },
+      async ({ redisOptions }) => {
+        const processed: 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 () => 1,
+              defaultLimit: 1,
+            },
+          ],
+          startConsumers: false,
+        });
+
+        const redis = createRedisClient(redisOptions);
+
+        queue.onMessage(async (ctx) => {
+          if (ctx.message.payload.value === "msg-0") {
+            await redis.hdel(keys.inflightDataKey(0), ctx.message.id);
+          }
+          processed.push(ctx.message.payload.value);
+          await ctx.complete();
+        });
+
+        for (let i = 0; i < 2; i++) {
+          await queue.enqueue({
+            queueId: "tenant:t1:queue:q1",
+            tenantId: "t1",
+            payload: { value: `msg-${i}` },
+          });
+        }
+
+        queue.start();
+
+        await vi.waitFor(
+          () => {
+            expect(processed).toHaveLength(2);
+          },
+          { timeout: 10000 }
+        );
+
+        const held = await redis.scard(keys.concurrencyKey("tenant", "t1"));
+        expect(held).toBe(0);
+
+        await redis.quit();
+        await queue.close();
+      }
+    );
+  });
 });