fix(redis-worker): bound the reconcile sweep cost and write the release note for users

The sweep now scans with a larger COUNT, checks set members in fixed-size
chunks so one script call cannot hold up Redis on a set full of leaked
members, jitters its first run so instances sharing a Redis do not all sweep
at once, and is disabled entirely by a zero interval instead of spinning.
The changeset now leads with the user-visible fix instead of the mechanism.

Author: matt-aitken

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

@@ -2,4 +2,4 @@
 "@trigger.dev/redis-worker": patch
 ---
 
-Fair queue consumers no longer leak the concurrency slots that gate a tenant's throughput, and leaked slots now heal themselves. Slots are freed on the completion, retry, dead-letter, and reclaim paths that previously skipped them, and a failed release never blocks the message's own state transition, so a Redis error can no longer turn into a duplicate execution or a lost retry. A periodic reconcile loop removes any slot whose message is no longer in flight, and a message that still holds its own slot from an earlier failed release is re-admitted instead of being blocked by it.
+Fair queue tenants can no longer get permanently stuck behind leaked concurrency slots. Slots are now freed on every path that finishes a message, a failed release no longer causes a message to run twice or lose its retry, and a background sweep frees any slot that does leak, so a tenant's queues recover on their own instead of needing manual cleanup.

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

@@ -7,6 +7,12 @@ import type {
   QueueDescriptor,
 } from "./types.js";
 
+/**
+ * Caps how many set members a single sweep script invocation checks, bounding the time the
+ * atomic Lua script can hold up Redis when a set has accumulated many leaked members.
+ */
+const SWEEP_MEMBER_CHUNK_SIZE = 500;
+
 export interface ConcurrencyManagerOptions {
   redis: RedisOptions;
   keys: FairQueueKeyProducer;
@@ -198,7 +204,7 @@ export class ConcurrencyManager {
           "MATCH",
           pattern,
           "COUNT",
-          100
+          1000
         );
         cursor = nextCursor;
 
@@ -213,12 +219,15 @@ export class ConcurrencyManager {
             }
             scannedSets++;
 
-            const removedIds = await this.redis.removeOrphanedConcurrencySlots(
-              1 + inflightDataKeys.length,
-              [key, ...inflightDataKeys],
-              ...members
-            );
-            removed.push(...removedIds);
+            for (let i = 0; i < members.length; i += SWEEP_MEMBER_CHUNK_SIZE) {
+              const chunk = members.slice(i, i + SWEEP_MEMBER_CHUNK_SIZE);
+              const removedIds = await this.redis.removeOrphanedConcurrencySlots(
+                1 + inflightDataKeys.length,
+                [key, ...inflightDataKeys],
+                ...chunk
+              );
+              removed.push(...removedIds);
+            }
           } catch (error) {
             this.logger.error("Failed to sweep concurrency set, skipping it", {
               key,

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

@@ -2,7 +2,7 @@ import { createRedisClient, type Redis } from "@internal/redis";
 import { SpanKind, type Span } from "@internal/tracing";
 import { Logger } from "@trigger.dev/core/logger";
 import { nanoid } from "nanoid";
-import { setInterval } from "node:timers/promises";
+import { setInterval, setTimeout as delay } from "node:timers/promises";
 import { type z } from "zod";
 import { isAbortError } from "../utils.js";
 import { ConcurrencyManager } from "./concurrency.js";
@@ -665,7 +665,7 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
     // Start reclaim loop for handling timed-out messages
     this.reclaimLoop = this.#runReclaimLoop();
 
-    if (this.concurrencyManager) {
+    if (this.concurrencyManager && this.reconcileIntervalMs > 0) {
       this.reconcileLoop = this.#runReconcileLoop();
     }
 
@@ -1650,8 +1650,15 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
     }
   }
 
+  /**
+   * The initial delay is jittered so instances sharing a Redis do not all sweep at once.
+   */
   async #runReconcileLoop(): Promise<void> {
     try {
+      await delay(Math.random() * this.reconcileIntervalMs, null, {
+        signal: this.abortController.signal,
+      });
+
       for await (const _ of setInterval(this.reconcileIntervalMs, null, {
         signal: this.abortController.signal,
       })) {

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

@@ -803,13 +803,15 @@ describe("ConcurrencyManager", () => {
         const inflightDataKey = keys.inflightDataKey(0);
 
         try {
-          await redis.sadd(concurrencyKey, "orphan-1", "orphan-2", "active-1");
+          const orphans = Array.from({ length: 1200 }, (_, i) => `orphan-${i}`);
+          await redis.sadd(concurrencyKey, ...orphans, "active-1", "active-2");
           await redis.hset(inflightDataKey, "active-1", "{}");
+          await redis.hset(inflightDataKey, "active-2", "{}");
 
           const result = await manager.sweepOrphanedSlots([inflightDataKey]);
 
-          expect(result.removed.sort()).toEqual(["orphan-1", "orphan-2"]);
-          expect(await redis.smembers(concurrencyKey)).toEqual(["active-1"]);
+          expect(result.removed.sort()).toEqual(orphans.sort());
+          expect((await redis.smembers(concurrencyKey)).sort()).toEqual(["active-1", "active-2"]);
         } finally {
           await redis.del(concurrencyKey);
           await redis.del(inflightDataKey);

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

@@ -2118,5 +2118,77 @@ describe("FairQueue", () => {
         }
       }
     );
+
+    redisTest(
+      "should not sweep when the reconcile interval is zero",
+      { timeout: 10000 },
+      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: 60000,
+          reclaimIntervalMs: 100,
+          reconcileIntervalMs: 0,
+          concurrencyGroups: [
+            {
+              name: "tenant",
+              extractGroupId: (q) => q.tenantId,
+              getLimit: async () => 5,
+              defaultLimit: 5,
+            },
+          ],
+          startConsumers: false,
+        });
+
+        const redis = createRedisClient(redisOptions);
+        const queueId = "tenant:t1:queue:sweep-disabled";
+        const concurrencyKey = keys.concurrencyKey("tenant", "t1");
+        const processed: string[] = [];
+
+        try {
+          await redis.sadd(concurrencyKey, "ghost-message");
+
+          queue.onMessage(async (ctx) => {
+            processed.push(ctx.message.payload.value);
+            await ctx.complete();
+          });
+
+          await queue.enqueue({
+            queueId,
+            tenantId: "t1",
+            payload: { value: "msg-0" },
+          });
+
+          queue.start();
+
+          await vi.waitFor(
+            () => {
+              expect(processed).toEqual(["msg-0"]);
+            },
+            { timeout: 5000 }
+          );
+
+          await new Promise((resolve) => setTimeout(resolve, 1000));
+
+          expect(await redis.sismember(concurrencyKey, "ghost-message")).toBe(1);
+        } finally {
+          await redis.del(concurrencyKey);
+          await redis.quit();
+          await queue.close();
+        }
+      }
+    );
   });
 });

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

@@ -413,7 +413,7 @@ export interface FairQueueOptions<TPayloadSchema extends z.ZodTypeAny = z.ZodUnk
   heartbeatIntervalMs?: number;
   /** Interval for reclaiming timed-out messages (default: 5000) */
   reclaimIntervalMs?: number;
-  /** Interval for sweeping orphaned concurrency slots (default: 60000) */
+  /** Interval for sweeping orphaned concurrency slots; 0 disables the sweep (default: 60000) */
   reconcileIntervalMs?: number;
 
   // Consumers