fix(redis-worker,run-engine): give the item heartbeat real margin and stop one bad key stalling a shard

Each beat extended the deadline by the same amount as the gap between beats, so
timer drift and round-trip latency left the message briefly past its deadline on
every cycle, and a reclaim scan landing in that window still handed a running item
to a second consumer. Beats now extend by a full visibility timeout while the tick
stays at a third of it.

Reclaim released the whole timed-out batch in one pipeline and aborted every
requeue if any command failed, so a single unusable concurrency key stalled reclaim
for every other tenant in that shard. A failed batch now falls back to releasing
message by message and only the messages whose slot could not be freed are held
back.

Author: matt-aitken

internal-packages/run-engine/src/batch-queue/index.ts

@@ -162,6 +162,7 @@ export class BatchQueue {
       consumerCount: options.consumerCount,
       consumerIntervalMs: options.consumerIntervalMs,
       visibilityTimeoutMs: this.visibilityTimeoutMs,
+      heartbeatIntervalMs: this.visibilityTimeoutMs,
       startConsumers: false, // We control when to start
       cooloff: {
         enabled: false,

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

@@ -1550,21 +1550,45 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
    * leaves the messages in-flight for the next reclaim tick. That is the safe direction:
    * requeuing a message whose slot is still held is what strands the slot permanently.
    */
-  async #releaseReclaimedConcurrency(messages: ReclaimedMessageInfo[]): Promise<void> {
+  async #releaseReclaimedConcurrency(messages: ReclaimedMessageInfo[]): Promise<string[]> {
     if (!this.concurrencyManager || messages.length === 0) {
-      return;
+      return [];
     }
 
-    await this.concurrencyManager.releaseBatch(
-      messages.map((message) => ({
-        queue: {
-          id: message.queueId,
-          tenantId: message.tenantId,
-          metadata: message.metadata ?? {},
-        },
-        messageId: message.messageId,
-      }))
-    );
+    const descriptorFor = (message: ReclaimedMessageInfo) => ({
+      id: message.queueId,
+      tenantId: message.tenantId,
+      metadata: message.metadata ?? {},
+    });
+
+    try {
+      await this.concurrencyManager.releaseBatch(
+        messages.map((message) => ({ queue: descriptorFor(message), messageId: message.messageId }))
+      );
+      return [];
+    } catch (error) {
+      this.logger.error("Batch concurrency release failed, retrying message by message", {
+        count: messages.length,
+        error: error instanceof Error ? error.message : String(error),
+      });
+    }
+
+    const failed: string[] = [];
+
+    for (const message of messages) {
+      try {
+        await this.concurrencyManager.release(descriptorFor(message), message.messageId);
+      } catch (error) {
+        failed.push(message.messageId);
+        this.logger.error("Failed to release concurrency for reclaimed message", {
+          messageId: message.messageId,
+          queueId: message.queueId,
+          error: error instanceof Error ? error.message : String(error),
+        });
+      }
+    }
+
+    return failed;
   }
 
   async #reclaimTimedOutMessages(): Promise<void> {

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

@@ -1580,6 +1580,16 @@ describe("FairQueue", () => {
 
           expect(await redis.zcard(keys.inflightKey(0))).toBe(1);
           expect(await redis.zcard(keys.queueKey(queueId))).toBe(0);
+
+          await redis.del(concurrencyKey);
+
+          await vi.waitFor(
+            async () => {
+              expect(await redis.zcard(keys.queueKey(queueId))).toBe(1);
+              expect(await redis.zcard(keys.inflightKey(0))).toBe(0);
+            },
+            { timeout: 8000 }
+          );
         } finally {
           await redis.del(concurrencyKey);
           await redis.quit();

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

@@ -399,7 +399,7 @@ export class VisibilityManager {
       dispatchKey: string;
       tenantId: string;
     },
-    onBeforeRequeue?: (messages: ReclaimedMessageInfo[]) => Promise<void>
+    onBeforeRequeue?: (messages: ReclaimedMessageInfo[]) => Promise<string[] | void>
   ): Promise<ReclaimedMessageInfo[]> {
     const inflightKey = this.keys.inflightKey(shardId);
     const inflightDataKey = this.keys.inflightDataKey(shardId);
@@ -465,14 +465,24 @@ export class VisibilityManager {
       return [];
     }
 
-    if (onBeforeRequeue) {
-      await onBeforeRequeue(candidates.map((candidate) => candidate.info));
-    }
+    const notReleased = new Set(
+      (onBeforeRequeue
+        ? await onBeforeRequeue(candidates.map((candidate) => candidate.info))
+        : undefined) ?? []
+    );
 
     const reclaimedMessages: ReclaimedMessageInfo[] = [];
 
     for (const { member, deadlineScore, storedMessage, info } of candidates) {
       const { messageId, queueId } = info;
+
+      if (notReleased.has(messageId)) {
+        this.logger.error("Skipping requeue, concurrency slot was not released", {
+          messageId,
+          queueId,
+        });
+        continue;
+      }
       const { queueKey, queueItemsKey, tenantQueueIndexKey, dispatchKey, tenantId } =
         getQueueKeys(queueId);