fix(run-engine,redis-worker): drop a batch item result once its lease is lost

The heartbeat already learns when another consumer has taken the item over: the
in-flight member is gone, so the extend returns false. That signal was discarded,
leaving the original consumer free to finish and complete over the new owner's
in-flight record. It now stops and discards its result instead, so the consumer
that actually owns the item is the one that finishes it.

Also treat a discarded release pipeline as a failure rather than a success, since
a null result means none of the commands ran.

Author: matt-aitken

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

@@ -767,20 +767,32 @@ export class BatchQueue {
    * 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 {
+  #startHeartbeat(
+    messageId: string,
+    queueId: string
+  ): { stop: () => void; lostLease: () => boolean } {
+    let lostLease = false;
+
     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.fairQueue
+        .heartbeatMessage(messageId, queueId)
+        .then((stillOwned) => {
+          if (!stillOwned) {
+            lostLease = true;
+          }
+        })
+        .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);
+    return { stop: () => clearInterval(interval), lostLease: () => lostLease };
   }
 
   async #handleMessage(consumerId: string, messageId: string, queueId: string): Promise<void> {
@@ -851,7 +863,7 @@ export class BatchQueue {
       let processedCount: number;
 
       try {
-        const stopHeartbeat = this.#startHeartbeat(messageId, queueId);
+        const heartbeat = this.#startHeartbeat(messageId, queueId);
         let result: Awaited<ReturnType<ProcessBatchItemCallback>>;
         try {
           result = await this.#startSpan("BatchQueue.processItemCallback", async (innerSpan) => {
@@ -871,7 +883,17 @@ export class BatchQueue {
             });
           });
         } finally {
-          stopHeartbeat();
+          heartbeat.stop();
+        }
+
+        if (heartbeat.lostLease()) {
+          this.logger.warn("Discarding batch item result, another consumer now owns it", {
+            batchId,
+            itemIndex,
+            messageId,
+            attempt,
+          });
+          return;
         }
 
         if (result.success) {

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

@@ -115,7 +115,13 @@ export class ConcurrencyManager {
     results: Array<[Error | null, unknown]> | null,
     messageCount: number
   ): void {
-    const errors = (results ?? [])
+    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));