fix(redis-worker): heal leaked concurrency slots instead of blocking on release failures

A failed slot release previously aborted its caller: completeMessage left the
message in flight to be re-delivered as a duplicate execution, the retry path
lost the attempt increment, and the reclaim path held messages in flight
indefinitely. Release is now best-effort on every path: the primary state
transition proceeds and the failure is logged.

The resulting leaks are recoverable in two ways. reserve re-admits a message
that already holds its own slot, since re-admission does not increase
concurrency. A reconcile loop removes any slot member with no in-flight
record; the check-and-remove is atomic and sound because a message is always
registered in flight before its slot is reserved.

Author: matt-aitken

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

@@ -2,6 +2,4 @@
 "@trigger.dev/redis-worker": patch
 ---
 
-Fair queue consumers no longer leak the concurrency slots that gate a tenant's throughput. Slots were held by messages that had already finished, were never reclaimed, and once enough of them accumulated every queue belonging to that tenant stopped being served. Slots are now freed on the paths that previously skipped them, freed before the record needed to recover them is discarded, and released before a reclaimed message goes back on the queue. A failed release is now surfaced instead of being silently treated as success.
-
-Concurrency groups keyed on queue metadata rather than the tenant can still resolve to the wrong group when a consumer completes a message it did not enqueue, so this does not yet cover that case.
+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.

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

@@ -11,6 +11,10 @@ export interface ConcurrencyManagerOptions {
   redis: RedisOptions;
   keys: FairQueueKeyProducer;
   groups: ConcurrencyGroupConfig[];
+  logger?: {
+    debug: (message: string, context?: Record<string, unknown>) => void;
+    error: (message: string, context?: Record<string, unknown>) => void;
+  };
 }
 
 /**
@@ -26,12 +30,17 @@ export class ConcurrencyManager {
   private keys: FairQueueKeyProducer;
   private groups: ConcurrencyGroupConfig[];
   private groupsByName: Map<string, ConcurrencyGroupConfig>;
+  private logger: NonNullable<ConcurrencyManagerOptions["logger"]>;
 
   constructor(private options: ConcurrencyManagerOptions) {
     this.redis = createRedisClient(options.redis);
     this.keys = options.keys;
     this.groups = options.groups;
     this.groupsByName = new Map(options.groups.map((g) => [g.name, g]));
+    this.logger = options.logger ?? {
+      debug: () => {},
+      error: () => {},
+    };
 
     this.#registerCommands();
   }
@@ -160,6 +169,69 @@ export class ConcurrencyManager {
     this.#assertPipelineSucceeded(await pipeline.exec(), messages.length);
   }
 
+  /**
+   * Remove concurrency set members that no longer correspond to an in-flight message,
+   * healing slots leaked by failed releases. Scans every set of every group and, per
+   * member, atomically removes it unless the message id appears in one of the given
+   * in-flight data hashes. Sound because a message is registered in-flight before its
+   * slot is reserved, so at the moment of the atomic check a member with no in-flight
+   * record can only be a leak; if the message is about to be re-claimed, reserve simply
+   * re-adds the member.
+   *
+   * @param inflightDataKeys - The in-flight data hash keys for every shard
+   * @returns The message ids that were removed, and how many sets were checked
+   */
+  async sweepOrphanedSlots(
+    inflightDataKeys: string[]
+  ): Promise<{ scannedSets: number; removed: string[] }> {
+    const keyPrefix = this.options.redis.keyPrefix ?? "";
+    let scannedSets = 0;
+    const removed: string[] = [];
+
+    for (const group of this.groups) {
+      const pattern = `${keyPrefix}${this.keys.concurrencyKey(group.name, "*")}`;
+      let cursor = "0";
+
+      do {
+        const [nextCursor, foundKeys] = await this.redis.scan(
+          cursor,
+          "MATCH",
+          pattern,
+          "COUNT",
+          100
+        );
+        cursor = nextCursor;
+
+        for (const fullKey of foundKeys) {
+          const key =
+            keyPrefix && fullKey.startsWith(keyPrefix) ? fullKey.slice(keyPrefix.length) : fullKey;
+
+          try {
+            const members = await this.redis.smembers(key);
+            if (members.length === 0) {
+              continue;
+            }
+            scannedSets++;
+
+            const removedIds = await this.redis.removeOrphanedConcurrencySlots(
+              1 + inflightDataKeys.length,
+              [key, ...inflightDataKeys],
+              ...members
+            );
+            removed.push(...removedIds);
+          } catch (error) {
+            this.logger.error("Failed to sweep concurrency set, skipping it", {
+              key,
+              error: error instanceof Error ? error.message : String(error),
+            });
+          }
+        }
+      } while (cursor !== "0");
+    }
+
+    return { scannedSets, removed };
+  }
+
   /**
    * Get current concurrency for a specific group.
    */
@@ -298,14 +370,20 @@ export class ConcurrencyManager {
 local numGroups = #KEYS
 local messageId = ARGV[1]
 
--- Check all groups first
+-- Check all groups first. A message that is already a member of a group's set passes
+-- that group's check: re-admitting it does not increase concurrency (SADD is a no-op),
+-- and counting its own leftover slot against it would let a message whose earlier
+-- release failed block its own retry forever.
 for i = 1, numGroups do
   local key = KEYS[i]
   local limit = tonumber(ARGV[1 + i])  -- Limits start at ARGV[2]
-  local current = redis.call('SCARD', key)
-  
-  if current >= limit then
-    return 0  -- At capacity
+
+  if redis.call('SISMEMBER', key, messageId) == 0 then
+    local current = redis.call('SCARD', key)
+
+    if current >= limit then
+      return 0  -- At capacity
+    end
   end
 end
 
@@ -318,6 +396,37 @@ end
 return 1
       `,
     });
+
+    // Atomic orphan sweep for one concurrency set
+    // KEYS[1]: concurrency set key
+    // KEYS[2..n]: in-flight data hash keys for every shard
+    // ARGV: candidate messageIds (a snapshot of the set's members)
+    this.redis.defineCommand("removeOrphanedConcurrencySlots", {
+      lua: `
+local concurrencyKey = KEYS[1]
+local removedIds = {}
+
+for i = 1, #ARGV do
+  local messageId = ARGV[i]
+  local inflight = false
+
+  for j = 2, #KEYS do
+    if redis.call('HEXISTS', KEYS[j], messageId) == 1 then
+      inflight = true
+      break
+    end
+  end
+
+  if not inflight then
+    if redis.call('SREM', concurrencyKey, messageId) == 1 then
+      table.insert(removedIds, messageId)
+    end
+  end
+end
+
+return removedIds
+      `,
+    });
   }
 }
 
@@ -330,5 +439,11 @@ declare module "@internal/redis" {
       messageId: string,
       ...limits: string[]
     ): Promise<number>;
+
+    removeOrphanedConcurrencySlots(
+      numKeys: number,
+      keys: string[],
+      ...messageIds: string[]
+    ): Promise<string[]>;
   }
 }