fix(redis-worker): keep metadata-derived concurrency groups on the release path

Concurrency groups can key on queue metadata, not just the tenant. The fallback
descriptor dropped metadata and skipped the descriptor cache, so those groups
released against the wrong Redis set and leaked exactly as before. Prefer the
cached descriptor in both branches, and move the retry path's release ahead of
the re-queue so a redelivery cannot have its fresh reservation deleted by the
previous holder.

Author: matt-aitken

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

@@ -1245,13 +1245,11 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
       }
     }
 
-    const descriptor: QueueDescriptor = storedMessage
-      ? (this.queueDescriptorCache.get(queueId) ?? {
-          id: queueId,
-          tenantId: storedMessage.tenantId,
-          metadata: storedMessage.metadata ?? {},
-        })
-      : { id: queueId, tenantId: this.keys.extractTenantId(queueId), metadata: {} };
+    const descriptor: QueueDescriptor = this.queueDescriptorCache.get(queueId) ?? {
+      id: queueId,
+      tenantId: storedMessage?.tenantId ?? this.keys.extractTenantId(queueId),
+      metadata: storedMessage?.metadata ?? {},
+    };
 
     // Release concurrency
     if (this.concurrencyManager) {
@@ -1300,13 +1298,11 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
       }
     }
 
-    const descriptor: QueueDescriptor = storedMessage
-      ? (this.queueDescriptorCache.get(queueId) ?? {
-          id: queueId,
-          tenantId: storedMessage.tenantId,
-          metadata: storedMessage.metadata ?? {},
-        })
-      : { id: queueId, tenantId: this.keys.extractTenantId(queueId), metadata: {} };
+    const descriptor: QueueDescriptor = this.queueDescriptorCache.get(queueId) ?? {
+      id: queueId,
+      tenantId: storedMessage?.tenantId ?? this.keys.extractTenantId(queueId),
+      metadata: storedMessage?.metadata ?? {},
+    };
 
     // Release concurrency
     if (this.concurrencyManager) {
@@ -1411,6 +1407,11 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
           attempt: storedMessage.attempt + 1,
         };
 
+        // Release concurrency
+        if (this.concurrencyManager) {
+          await this.concurrencyManager.release(descriptor, storedMessage.id);
+        }
+
         // Release with delay, passing the updated message data so the Lua script
         // atomically writes the incremented attempt count when re-queuing.
         const tenantQueueIndexKey = this.keys.tenantQueueIndexKey(descriptor.tenantId);
@@ -1427,11 +1428,6 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
           JSON.stringify(updatedMessage)
         );
 
-        // Release concurrency
-        if (this.concurrencyManager) {
-          await this.concurrencyManager.release(descriptor, storedMessage.id);
-        }
-
         this.telemetry.recordRetry();
 
         this.logger.debug("Message scheduled for retry", {

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

@@ -1406,36 +1406,114 @@ describe("FairQueue", () => {
 
         const redis = createRedisClient(redisOptions);
 
-        queue.onMessage(async (ctx) => {
-          if (ctx.message.payload.value === "msg-0") {
-            await redis.hdel(keys.inflightDataKey(0), ctx.message.id);
+        try {
+          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}` },
+            });
           }
-          processed.push(ctx.message.payload.value);
-          await ctx.complete();
+
+          queue.start();
+
+          await vi.waitFor(
+            () => {
+              expect(processed).toHaveLength(2);
+            },
+            { timeout: 10000 }
+          );
+
+          const held = await redis.scard(keys.concurrencyKey("tenant", "t1"));
+          expect(held).toBe(0);
+        } finally {
+          await redis.quit();
+          await queue.close();
+        }
+      }
+    );
+
+    redisTest(
+      "should release metadata-derived concurrency groups when the in-flight record is gone",
+      { timeout: 15000 },
+      async ({ redisOptions }) => {
+        const processed: string[] = [];
+        keys = new DefaultFairQueueKeyProducer({ prefix: "test" });
+
+        const scheduler = new DRRScheduler({
+          redis: redisOptions,
+          keys,
+          quantum: 10,
+          maxDeficit: 100,
         });
 
-        for (let i = 0; i < 2; i++) {
-          await queue.enqueue({
-            queueId: "tenant:t1:queue:q1",
-            tenantId: "t1",
-            payload: { value: `msg-${i}` },
+        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,
+            },
+            {
+              name: "organization",
+              extractGroupId: (q) => (q.metadata.orgId as string) ?? "default",
+              getLimit: async () => 1,
+              defaultLimit: 1,
+            },
+          ],
+          startConsumers: false,
+        });
+
+        const redis = createRedisClient(redisOptions);
+
+        try {
+          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();
           });
-        }
 
-        queue.start();
+          for (let i = 0; i < 2; i++) {
+            await queue.enqueue({
+              queueId: "tenant:t1:queue:q1",
+              tenantId: "t1",
+              metadata: { orgId: "org-1" },
+              payload: { value: `msg-${i}` },
+            });
+          }
 
-        await vi.waitFor(
-          () => {
-            expect(processed).toHaveLength(2);
-          },
-          { timeout: 10000 }
-        );
+          queue.start();
 
-        const held = await redis.scard(keys.concurrencyKey("tenant", "t1"));
-        expect(held).toBe(0);
+          await vi.waitFor(
+            () => {
+              expect(processed).toHaveLength(2);
+            },
+            { timeout: 10000 }
+          );
 
-        await redis.quit();
-        await queue.close();
+          expect(await redis.scard(keys.concurrencyKey("organization", "org-1"))).toBe(0);
+          expect(await redis.scard(keys.concurrencyKey("organization", "default"))).toBe(0);
+        } finally {
+          await redis.quit();
+          await queue.close();
+        }
       }
     );
   });