Added fallback to extract tenantId from queueId when message data is missing or corrupted

Renamed misleading originalScore variable to _deadlineScore (it's the visibility deadline, not original timestamp)

Author: ericallam

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

@@ -659,6 +659,11 @@ describe("VisibilityManager", () => {
         const queueCount = await redis.zcard(queueKey);
         expect(queueCount).toBe(1);
 
+        // Verify message is back in queue with its original timestamp (not the deadline)
+        const queueMessages = await redis.zrange(queueKey, 0, -1, "WITHSCORES");
+        expect(queueMessages[0]).toBe(messageId);
+        expect(parseInt(queueMessages[1]!)).toBe(storedMessage.timestamp);
+
         // Verify message is no longer in-flight
         const inflightCount = await manager.getTotalInflightCount();
         expect(inflightCount).toBe(0);
@@ -775,6 +780,71 @@ describe("VisibilityManager", () => {
         await redis.quit();
       }
     );
+
+    redisTest(
+      "should use fallback tenantId extraction when message data is missing or corrupted",
+      { timeout: 10000 },
+      async ({ redisOptions }) => {
+        keys = new DefaultFairQueueKeyProducer({ prefix: "test" });
+
+        const manager = new VisibilityManager({
+          redis: redisOptions,
+          keys,
+          shardCount: 1,
+          defaultTimeoutMs: 100,
+        });
+
+        const redis = createRedisClient(redisOptions);
+        const queueId = "tenant:t1:queue:fallback-test";
+        const queueKey = keys.queueKey(queueId);
+        const queueItemsKey = keys.queueItemsKey(queueId);
+        const masterQueueKey = keys.masterQueueKey(0);
+        const inflightDataKey = keys.inflightDataKey(0);
+
+        // Add and claim a message
+        const messageId = "fallback-msg";
+        const storedMessage = {
+          id: messageId,
+          queueId,
+          tenantId: "t1",
+          payload: { id: 1 },
+          timestamp: Date.now() - 1000,
+          attempt: 1,
+          metadata: { orgId: "org-123" },
+        };
+
+        await redis.zadd(queueKey, storedMessage.timestamp, messageId);
+        await redis.hset(queueItemsKey, messageId, JSON.stringify(storedMessage));
+
+        // Claim the message
+        const claimResult = await manager.claim(queueId, queueKey, queueItemsKey, "consumer-1", 100);
+        expect(claimResult.claimed).toBe(true);
+
+        // Corrupt the in-flight data by setting invalid JSON
+        await redis.hset(inflightDataKey, messageId, "not-valid-json{{{");
+
+        // Wait for timeout
+        await new Promise((resolve) => setTimeout(resolve, 150));
+
+        // Reclaim should still work using fallback extraction
+        const reclaimedMessages = await manager.reclaimTimedOut(0, (qId) => ({
+          queueKey: keys.queueKey(qId),
+          queueItemsKey: keys.queueItemsKey(qId),
+          masterQueueKey,
+        }));
+
+        expect(reclaimedMessages).toHaveLength(1);
+        expect(reclaimedMessages[0]).toEqual({
+          messageId,
+          queueId,
+          tenantId: "t1", // Extracted from queueId via fallback
+          metadata: {}, // Empty metadata since we couldn't parse the stored message
+        });
+
+        await manager.close();
+        await redis.quit();
+      }
+    );
   });
 });
 

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

@@ -45,8 +45,8 @@ export class VisibilityManager {
     this.shardCount = options.shardCount;
     this.defaultTimeoutMs = options.defaultTimeoutMs;
     this.logger = options.logger ?? {
-      debug: () => {},
-      error: () => {},
+      debug: () => { },
+      error: () => { },
     };
 
     this.#registerCommands();
@@ -403,8 +403,8 @@ export class VisibilityManager {
 
     for (let i = 0; i < timedOut.length; i += 2) {
       const member = timedOut[i];
-      const originalScore = timedOut[i + 1];
-      if (!member || !originalScore) {
+      const _deadlineScore = timedOut[i + 1]; // This is the visibility deadline, not the original timestamp
+      if (!member || !_deadlineScore) {
         continue;
       }
       const { messageId, queueId } = this.#parseMember(member);
@@ -422,8 +422,9 @@ export class VisibilityManager {
           }
         }
 
-        // Re-add to queue with original score (or now if not available)
-        const score = parseFloat(originalScore) || now;
+        // Re-add to queue with original timestamp to preserve priority
+        // Fall back to now if we can't get the original timestamp
+        const score = storedMessage?.timestamp ?? now;
         await this.redis.releaseMessage(
           inflightKey,
           inflightDataKey,
@@ -437,19 +438,33 @@ export class VisibilityManager {
         );
 
         // Track reclaimed message for concurrency release
+        // Always add to reclaimedMessages to avoid concurrency leaks
         if (storedMessage) {
           reclaimedMessages.push({
             messageId,
             queueId,
             tenantId: storedMessage.tenantId,
             metadata: storedMessage.metadata,
           });
+        } else {
+          // Fallback: extract tenantId from queueId when message data is missing or corrupted
+          // This ensures concurrency is released even if we can't get the full metadata
+          this.logger.error("Missing or corrupted message data during reclaim, using fallback", {
+            messageId,
+            queueId,
+          });
+          reclaimedMessages.push({
+            messageId,
+            queueId,
+            tenantId: this.keys.extractTenantId(queueId),
+            metadata: {},
+          });
         }
 
         this.logger.debug("Reclaimed timed-out message", {
           messageId,
           queueId,
-          originalScore,
+          deadline: _deadlineScore,
         });
       } catch (error) {
         this.logger.error("Failed to reclaim message", {