fix(redis-worker,run-engine): close the remaining ways a batch item is stranded or run twice

Releasing a slot could silently do nothing. The release pipeline never inspected
its per-command errors, and ioredis resolves a pipeline whose commands failed, so
a failed SREM reported success and the caller went on to destroy the in-flight
record. Release now throws on a failed command.

Reclaim then acted on that false success: it released per message, swallowed any
error, and requeued regardless, which is exactly how a slot ends up held with the
message gone. It now frees the whole timed-out batch in one pipeline before any of
them is requeued, and a failure aborts the requeue so the messages stay in-flight
for the next tick. That also restores the single round trip per shard.

Requeuing removed the message from in-flight before writing it back to the queue.
Redis does not roll back a script that fails partway, so a failed queue write lost
the message outright. The order is now write-then-remove.

Batch items were never heartbeated, so any item slower than the visibility timeout
was reclaimed and handed to a second consumer while the first was still running it,
executing the item twice. Items are now heartbeated for as long as their callback
runs, and the visibility timeout is configurable so this is testable.

Author: matt-aitken

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

@@ -59,6 +59,9 @@ const ENV_CONCURRENCY_KEY_PREFIX = "batch:env_concurrency";
 // then all messages are routed to this queue for BatchQueue's own consumer loop.
 const BATCH_WORKER_QUEUE_ID = "batch-worker-queue";
 
+/** How long a claimed batch item stays invisible before the reclaim loop takes it back. */
+const BATCH_ITEM_VISIBILITY_TIMEOUT_MS = 60_000;
+
 export class BatchQueue {
   private fairQueue: FairQueue<typeof BatchItemPayloadSchema>;
   private workerQueueManager: WorkerQueueManager;
@@ -67,6 +70,8 @@ export class BatchQueue {
   private tracer?: Tracer;
   private concurrencyRedis: Redis;
   private defaultConcurrency: number;
+  private heartbeatIntervalMs: number;
+  private visibilityTimeoutMs: number;
   private maxAttempts: number;
 
   private processItemCallback?: ProcessBatchItemCallback;
@@ -95,6 +100,8 @@ export class BatchQueue {
     this.logger = options.logger ?? new Logger("BatchQueue", options.logLevel ?? "info");
     this.tracer = options.tracer;
     this.defaultConcurrency = options.defaultConcurrency ?? 10;
+    this.visibilityTimeoutMs = options.visibilityTimeoutMs ?? BATCH_ITEM_VISIBILITY_TIMEOUT_MS;
+    this.heartbeatIntervalMs = Math.max(50, Math.floor(this.visibilityTimeoutMs / 3));
     this.maxAttempts = options.retry?.maxAttempts ?? 1;
     this.abortController = new AbortController();
     this.workerQueueBlockingTimeoutSeconds = options.workerQueueBlockingTimeoutSeconds ?? 10;
@@ -154,7 +161,7 @@ export class BatchQueue {
       shardCount: options.shardCount ?? 1,
       consumerCount: options.consumerCount,
       consumerIntervalMs: options.consumerIntervalMs,
-      visibilityTimeoutMs: 60_000, // 1 minute for batch item processing
+      visibilityTimeoutMs: this.visibilityTimeoutMs,
       startConsumers: false, // We control when to start
       cooloff: {
         enabled: false,
@@ -752,6 +759,29 @@ export class BatchQueue {
   // Private - Message Handling
   // ============================================================================
 
+  /**
+   * Keep extending a message's visibility deadline while its callback runs, and return a
+   * function that stops doing so. Without this an item slower than the visibility timeout
+   * is redelivered while still being processed, and the original consumer's completion
+   * then destroys the redelivery's in-flight record, silently dropping the item and
+   * leaving the batch short of its expected count forever.
+   */
+  #startHeartbeat(messageId: string, queueId: string): () => void {
+    const interval = setInterval(() => {
+      this.fairQueue.heartbeatMessage(messageId, queueId).catch((error) => {
+        this.logger.debug("Batch item heartbeat failed", {
+          messageId,
+          queueId,
+          error: error instanceof Error ? error.message : String(error),
+        });
+      });
+    }, this.heartbeatIntervalMs);
+
+    interval.unref?.();
+
+    return () => clearInterval(interval);
+  }
+
   async #handleMessage(consumerId: string, messageId: string, queueId: string): Promise<void> {
     // Get message data from FairQueue's in-flight storage
     const storedMessage = await this.fairQueue.getMessageData(messageId, queueId);
@@ -820,9 +850,10 @@ export class BatchQueue {
       let processedCount: number;
 
       try {
-        const result = await this.#startSpan(
-          "BatchQueue.processItemCallback",
-          async (innerSpan) => {
+        const stopHeartbeat = this.#startHeartbeat(messageId, queueId);
+        let result: Awaited<ReturnType<ProcessBatchItemCallback>>;
+        try {
+          result = await this.#startSpan("BatchQueue.processItemCallback", async (innerSpan) => {
             innerSpan?.setAttributes({
               "batch.id": batchId,
               "batch.itemIndex": itemIndex,
@@ -837,8 +868,10 @@ export class BatchQueue {
               attempt,
               isFinalAttempt,
             });
-          }
-        );
+          });
+        } finally {
+          stopHeartbeat();
+        }
 
         if (result.success) {
           span?.setAttribute("batch.result", "success");

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

@@ -953,4 +953,56 @@ describe("BatchQueue", () => {
       }
     );
   });
+
+  describe("visibility heartbeat", () => {
+    redisTest(
+      "should not redeliver an item that takes longer than the visibility timeout",
+      { timeout: 60_000 },
+      async ({ redisContainer }) => {
+        const queue = new BatchQueue({
+          redis: {
+            host: redisContainer.getHost(),
+            port: redisContainer.getPort(),
+            keyPrefix: "test:",
+          },
+          drr: { quantum: 5, maxDeficit: 50 },
+          consumerCount: 2,
+          consumerIntervalMs: 50,
+          visibilityTimeoutMs: 1_000,
+          startConsumers: false,
+        });
+
+        const invocations: number[] = [];
+
+        try {
+          queue.onProcessItem(async ({ itemIndex }) => {
+            const isFirst = invocations.length === 0;
+            invocations.push(itemIndex);
+            if (isFirst) {
+              await new Promise((resolve) => setTimeout(resolve, 9_000));
+            }
+            return { success: true, runId: `run-${itemIndex}` };
+          });
+
+          await queue.initializeBatch(createInitOptions("batch-hb", "env-hb", 1));
+          await enqueueItems(queue, "batch-hb", "env-hb", createBatchItems(1));
+
+          queue.start();
+
+          await vi.waitFor(
+            () => {
+              expect(invocations.length).toBeGreaterThanOrEqual(1);
+            },
+            { timeout: 10_000 }
+          );
+
+          await new Promise((resolve) => setTimeout(resolve, 14_000));
+
+          expect(invocations).toEqual([0]);
+        } finally {
+          await queue.close();
+        }
+      }
+    );
+  });
 });

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

@@ -214,6 +214,12 @@ export type BatchQueueOptions = {
    * Items wait in queue until capacity frees up.
    */
   defaultConcurrency?: number;
+  /**
+   * How long a claimed item stays invisible before the reclaim loop takes it back.
+   * The item is heartbeated for as long as its callback runs, so this only bites when
+   * a consumer stops making progress. Defaults to 60s.
+   */
+  visibilityTimeoutMs?: number;
   /**
    * Optional global rate limiter to limit processing across all consumers.
    * When configured, limits the max items/second processed globally.

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

@@ -103,7 +103,31 @@ 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 {
+    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 +151,7 @@ export class ConcurrencyManager {
       }
     }
 
-    await pipeline.exec();
+    this.#assertPipelineSucceeded(await pipeline.exec(), messages.length);
   }
 
   /**

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

@@ -1498,28 +1498,14 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
       originalTimestamp: storedMessage.timestamp,
     };
 
+    // Complete in visibility manager
+    await this.visibilityManager.complete(storedMessage.id, storedMessage.queueId);
+
     // Add to DLQ
     const pipeline = this.redis.pipeline();
     pipeline.zadd(dlqKey, dlqMessage.deadLetteredAt, storedMessage.id);
     pipeline.hset(dlqDataKey, storedMessage.id, JSON.stringify(dlqMessage));
-    const dlqResults = await pipeline.exec();
-
-    const dlqErrors = (dlqResults ?? [])
-      .map(([error]) => error)
-      .filter((error): error is Error => Boolean(error));
-
-    if (dlqErrors.length > 0) {
-      this.logger.error("Failed to write message to DLQ, leaving it in-flight to be reclaimed", {
-        messageId: storedMessage.id,
-        queueId: storedMessage.queueId,
-        tenantId: storedMessage.tenantId,
-        errors: dlqErrors.map((error) => error.message),
-      });
-      return;
-    }
-
-    // Complete in visibility manager
-    await this.visibilityManager.complete(storedMessage.id, storedMessage.queueId);
+    await pipeline.exec();
 
     this.telemetry.recordDLQ();
 
@@ -1559,55 +1545,56 @@ export class FairQueue<TPayloadSchema extends z.ZodTypeAny = z.ZodUnknown> {
   }
 
   /**
-   * Free a reclaimed message's concurrency slot. Runs before the message is put back
-   * on the queue, so another consumer cannot reserve the same messageId and have its
-   * fresh reservation deleted by this release. Failures are logged rather than thrown
-   * so one bad message cannot stop the rest of the shard being reclaimed.
+   * Free every timed-out message's concurrency slot in one pipeline, before any of them
+   * is put back on the queue. Throwing here aborts the requeue for the whole batch, which
+   * 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(message: ReclaimedMessageInfo): Promise<void> {
-    if (!this.concurrencyManager) {
+  async #releaseReclaimedConcurrency(messages: ReclaimedMessageInfo[]): Promise<void> {
+    if (!this.concurrencyManager || messages.length === 0) {
       return;
     }
 
-    try {
-      await this.concurrencyManager.release(
-        {
+    await this.concurrencyManager.releaseBatch(
+      messages.map((message) => ({
+        queue: {
           id: message.queueId,
           tenantId: message.tenantId,
           metadata: message.metadata ?? {},
         },
-        message.messageId
-      );
-    } catch (error) {
-      this.logger.error("Failed to release concurrency for reclaimed message", {
         messageId: message.messageId,
-        queueId: message.queueId,
-        error: error instanceof Error ? error.message : String(error),
-      });
-    }
+      }))
+    );
   }
 
   async #reclaimTimedOutMessages(): Promise<void> {
     let totalReclaimed = 0;
 
     for (let shardId = 0; shardId < this.shardCount; shardId++) {
-      const reclaimedMessages = await this.visibilityManager.reclaimTimedOut(
-        shardId,
-        (queueId) => {
-          const tenantId = this.keys.extractTenantId(queueId);
-          const dispatchShardId = this.tenantDispatch.getShardForTenant(tenantId);
-          return {
-            queueKey: this.keys.queueKey(queueId),
-            queueItemsKey: this.keys.queueItemsKey(queueId),
-            tenantQueueIndexKey: this.keys.tenantQueueIndexKey(tenantId),
-            dispatchKey: this.keys.dispatchKey(dispatchShardId),
-            tenantId,
-          };
-        },
-        this.#releaseReclaimedConcurrency.bind(this)
-      );
+      try {
+        const reclaimedMessages = await this.visibilityManager.reclaimTimedOut(
+          shardId,
+          (queueId) => {
+            const tenantId = this.keys.extractTenantId(queueId);
+            const dispatchShardId = this.tenantDispatch.getShardForTenant(tenantId);
+            return {
+              queueKey: this.keys.queueKey(queueId),
+              queueItemsKey: this.keys.queueItemsKey(queueId),
+              tenantQueueIndexKey: this.keys.tenantQueueIndexKey(tenantId),
+              dispatchKey: this.keys.dispatchKey(dispatchShardId),
+              tenantId,
+            };
+          },
+          this.#releaseReclaimedConcurrency.bind(this)
+        );
 
-      totalReclaimed += reclaimedMessages.length;
+        totalReclaimed += reclaimedMessages.length;
+      } catch (error) {
+        this.logger.error("Failed to reclaim shard, leaving messages in-flight for the next tick", {
+          shardId,
+          error: error instanceof Error ? error.message : String(error),
+        });
+      }
     }
 
     if (totalReclaimed > 0) {