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

## Summary

Fair queue consumers could leak the per-tenant concurrency slots that
gate admission. Slots were freed on some paths and skipped on others,
and once enough leaked slots accumulated for a tenant, every queue that
tenant owned stopped being served until someone cleared the set by hand.
This PR frees slots on every path and, more importantly, makes the
remaining failure modes self-healing.

## Design

The fix applies one rule uniformly: releasing a concurrency slot is
best-effort cleanup and must never block the message's primary state
transition. Blocking completion re-delivers the message, which
duplicates customer work; blocking a retry loses the attempt increment,
so the message can circle forever; blocking a reclaim strands the
message in flight. A leaked slot is the better failure in every one of
those trades because it is the only one that is recoverable. A failed
release is therefore logged and the transition proceeds.

Leaked slots then heal through two mechanisms:

- `reserve` re-admits a message that is already a member of its own
concurrency set, since re-admitting it does not increase concurrency. A
message whose earlier release failed can no longer be blocked by its own
leftover slot.
- A reconcile loop periodically removes any set member with no in-flight
record (interval configurable via `reconcileIntervalMs`, default 60s).
The check-and-remove is atomic, and it is sound because a message is
always registered in flight before its slot is reserved, so a member
with no in-flight record can only be a leak. This also covers leaks this
PR cannot prevent directly, such as a release that resolves the wrong
concurrency group from queue metadata.

Ordering hardening from earlier revisions stays: slots are released
before the in-flight record needed to describe them is discarded, the
release Lua scripts write the message back to the queue before removing
it from in-flight (Lua does not roll back on error), and dangling
in-flight entries with no payload are dropped instead of being rescanned
forever.

Every guard test was verified to fail without its specific fix,
including the duplicate-execution case: completing a message while its
slot release fails used to re-deliver and re-execute it.

Author: matt-aitken

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

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/redis-worker": patch
+---
+
+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,10 +7,22 @@ import type {
   QueueDescriptor,
 } from "./types.js";
 
+/**
+ * Page size for iterating a concurrency set's members (SSCAN COUNT) and cap on how many
+ * members one sweep script invocation checks, bounding both the snapshot reads and 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;
   groups: ConcurrencyGroupConfig[];
+  logger?: {
+    debug: (message: string, context?: Record<string, unknown>) => void;
+    error: (message: string, context?: Record<string, unknown>) => void;
+  };
 }
 
 /**
@@ -26,12 +38,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();
   }
@@ -103,7 +120,37 @@ export class ConcurrencyManager {
       pipeline.srem(key, messageId);
     }
 
-    await pipeline.exec();
+    this.#assertPipelineSucceeded(await pipeline.exec(), 1);
+  }
+
+  /**
+   * Throw if any command in a released pipeline failed. ioredis resolves `exec()` even when
+   * individual commands error, so an unchecked pipeline reports success while leaving the
+   * slot held, which strands it permanently once the caller drops the in-flight record.
+   */
+  #assertPipelineSucceeded(
+    results: Array<[Error | null, unknown]> | null,
+    messageCount: number
+  ): void {
+    if (results === null) {
+      throw new Error(
+        `Concurrency release pipeline for ${messageCount} message(s) was discarded without executing`
+      );
+    }
+
+    const errors = results
+      .map(([error]) => error)
+      .filter((error): error is Error => Boolean(error));
+
+    if (errors.length > 0) {
+      throw new Error(
+        `Failed to release ${errors.length} of ${
+          results?.length ?? 0
+        } concurrency slot commands across ${messageCount} message(s): ${errors
+          .map((error) => error.message)
+          .join("; ")}`
+      );
+    }
   }
 
   /**
@@ -127,7 +174,98 @@ export class ConcurrencyManager {
       }
     }
 
-    await pipeline.exec();
+    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. The sweep
+   *   refuses to run when this is empty, since with nowhere to look for running messages
+   *   every member would look orphaned and all concurrency accounting would be erased.
+   * @returns The message ids that were removed, and how many sets were checked
+   */
+  async sweepOrphanedSlots(
+    inflightDataKeys: string[]
+  ): Promise<{ scannedSets: number; removed: string[] }> {
+    if (inflightDataKeys.length === 0) {
+      this.logger.error(
+        "Refusing to sweep concurrency slots without any in-flight data keys: every member would look orphaned and all concurrency accounting would be erased"
+      );
+      return { scannedSets: 0, removed: [] };
+    }
+
+    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",
+          1000
+        );
+        cursor = nextCursor;
+
+        for (const fullKey of foundKeys) {
+          const key =
+            keyPrefix && fullKey.startsWith(keyPrefix) ? fullKey.slice(keyPrefix.length) : fullKey;
+
+          try {
+            let sawMembers = false;
+            let memberCursor = "0";
+
+            do {
+              const [nextMemberCursor, page] = await this.redis.sscan(
+                key,
+                memberCursor,
+                "COUNT",
+                SWEEP_MEMBER_CHUNK_SIZE
+              );
+              memberCursor = nextMemberCursor;
+
+              if (page.length === 0) {
+                continue;
+              }
+              sawMembers = true;
+
+              for (let i = 0; i < page.length; i += SWEEP_MEMBER_CHUNK_SIZE) {
+                const chunk = page.slice(i, i + SWEEP_MEMBER_CHUNK_SIZE);
+                const removedIds = await this.redis.removeOrphanedConcurrencySlots(
+                  1 + inflightDataKeys.length,
+                  [key, ...inflightDataKeys],
+                  ...chunk
+                );
+                removed.push(...removedIds);
+              }
+            } while (memberCursor !== "0");
+
+            if (sawMembers) {
+              scannedSets++;
+            }
+          } 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 };
   }
 
   /**
@@ -268,14 +406,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
 
@@ -288,6 +432,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
+      `,
+    });
   }
 }
 
@@ -300,5 +475,11 @@ declare module "@internal/redis" {
       messageId: string,
       ...limits: string[]
     ): Promise<number>;
+
+    removeOrphanedConcurrencySlots(
+      numKeys: number,
+      keys: string[],
+      ...messageIds: string[]
+    ): Promise<string[]>;
   }
 }