fix(redis-worker): keep a failed attempt from overwriting, reviving or dead-lettering a newer enqueue

Fix a redis-worker job that was re-enqueued while a previous attempt was running being overwritten or moved to the dead letter queue when that earlier attempt failed.

Mono-RevId: 0fa23407e2e548cc0ee2841d301bbf0562ce8919

Author: 0ski

packages/redis-worker/src/queue.test.ts

@@ -603,4 +603,111 @@ describe("SimpleQueue", () => {
       }
     }
   );
+
+  redisTest(
+    "requeue and moveToDeadLetterQueue skip an item a newer enqueue replaced",
+    { timeout: 20_000 },
+    async ({ redisContainer }) => {
+      const queue = new SimpleQueue({
+        name: "test-supersede",
+        schema: { test: z.object({ value: z.number() }) },
+        redisOptions: {
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+          password: redisContainer.getPassword(),
+        },
+        logger: new Logger("test", "error"),
+      });
+
+      try {
+        await queue.enqueue({
+          id: "1",
+          job: "test",
+          item: { value: 1 },
+          visibilityTimeoutMs: 2000,
+        });
+        const [first] = await queue.dequeue(1);
+        await queue.enqueue({
+          id: "1",
+          job: "test",
+          item: { value: 2 },
+          visibilityTimeoutMs: 2000,
+        });
+
+        const retry = {
+          id: "1",
+          job: "test" as const,
+          item: { value: 1 },
+          attempt: 1,
+          availableAt: new Date(),
+          visibilityTimeoutMs: 2000,
+        };
+        expect(await queue.requeue({ ...retry, deduplicationKey: first?.deduplicationKey })).toBe(
+          false
+        );
+        expect(await queue.moveToDeadLetterQueue("1", "failed", first?.deduplicationKey)).toBe(
+          false
+        );
+        expect(await queue.sizeOfDeadLetterQueue()).toBe(0);
+
+        const [newer] = await queue.dequeue(1);
+        expect(newer).toEqual(expect.objectContaining({ id: "1", item: { value: 2 } }));
+
+        expect(await queue.requeue({ ...retry, deduplicationKey: newer?.deduplicationKey })).toBe(
+          true
+        );
+        const [retried] = await queue.dequeue(1);
+        expect(retried).toEqual(expect.objectContaining({ item: { value: 1 }, attempt: 1 }));
+        expect(await queue.moveToDeadLetterQueue("1", "failed", retried?.deduplicationKey)).toBe(
+          true
+        );
+        expect(await queue.sizeOfDeadLetterQueue()).toBe(1);
+      } finally {
+        await queue.close();
+      }
+    }
+  );
+
+  redisTest(
+    "requeue does not revive an item acked while the attempt ran",
+    { timeout: 20_000 },
+    async ({ redisContainer }) => {
+      const queue = new SimpleQueue({
+        name: "test-requeue-acked",
+        schema: { test: z.object({ value: z.number() }) },
+        redisOptions: {
+          host: redisContainer.getHost(),
+          port: redisContainer.getPort(),
+          password: redisContainer.getPassword(),
+        },
+        logger: new Logger("test", "error"),
+      });
+
+      try {
+        await queue.enqueue({
+          id: "1",
+          job: "test",
+          item: { value: 1 },
+          visibilityTimeoutMs: 2000,
+        });
+        const [first] = await queue.dequeue(1);
+        await queue.ack("1");
+
+        const requeued = await queue.requeue({
+          id: "1",
+          job: "test",
+          item: { value: 1 },
+          attempt: 1,
+          availableAt: new Date(),
+          visibilityTimeoutMs: 2000,
+          deduplicationKey: first?.deduplicationKey,
+        });
+
+        expect(requeued).toBe(false);
+        expect(await queue.getJob("1")).toBeNull();
+      } finally {
+        await queue.close();
+      }
+    }
+  );
 });

packages/redis-worker/src/queue.ts

@@ -140,6 +140,50 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
     }
   }
 
+  /**
+   * Re-enqueues an item for a retry, unless since it was dequeued a newer enqueue replaced it (its
+   * stored deduplication key no longer matches) or it was removed by an ack or cancel. Returns
+   * false when it skipped, so a failed attempt never overwrites the newer item or revives a removed
+   * one. A newer enqueue starts its own retry budget: an item a handler re-enqueues before failing
+   * restarts at attempt 0 rather than moving towards the DLQ.
+   */
+  async requeue({
+    id,
+    job,
+    item,
+    attempt,
+    availableAt,
+    visibilityTimeoutMs,
+    deduplicationKey,
+  }: {
+    id: string;
+    job: MessageCatalogKey<TMessageCatalog>;
+    item: MessageCatalogValue<TMessageCatalog, MessageCatalogKey<TMessageCatalog>>;
+    attempt: number;
+    availableAt: Date;
+    visibilityTimeoutMs: number;
+    deduplicationKey?: string;
+  }): Promise<boolean> {
+    const serializedItem = JSON.stringify({
+      job,
+      item,
+      visibilityTimeoutMs,
+      attempt,
+      deduplicationKey: nanoid(),
+    });
+
+    const result = await this.redis.requeueItem(
+      `queue`,
+      `items`,
+      id,
+      availableAt.getTime(),
+      serializedItem,
+      deduplicationKey ?? ""
+    );
+
+    return result === 1;
+  }
+
   async enqueueOnce({
     id,
     job,
@@ -362,7 +406,15 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
     };
   }
 
-  async moveToDeadLetterQueue(id: string, errorMessage: string): Promise<void> {
+  /**
+   * Moves an item to the dead letter queue. With a deduplication key, an item a newer enqueue has
+   * replaced is left in the queue and false is returned.
+   */
+  async moveToDeadLetterQueue(
+    id: string,
+    errorMessage: string,
+    deduplicationKey?: string
+  ): Promise<boolean> {
     try {
       this.logger.debug(`SimpleQueue ${this.name}.moveToDeadLetterQueue(): moving item to DLQ`, {
         queue: this.name,
@@ -376,12 +428,15 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
         `dlq`,
         `dlq:items`,
         id,
-        errorMessage
+        errorMessage,
+        deduplicationKey ?? ""
       );
 
+      if (result === 2) return false;
       if (result !== 1) {
         throw new Error("Move to Dead Letter Queue operation failed");
       }
+      return true;
     } catch (e) {
       this.logger.error(
         `SimpleQueue ${this.name}.moveToDeadLetterQueue(): error moving item to DLQ`,
@@ -596,6 +651,36 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
       `,
     });
 
+    this.redis.defineCommand("requeueItem", {
+      numberOfKeys: 2,
+      lua: `
+        local queue = KEYS[1]
+        local items = KEYS[2]
+        local id = ARGV[1]
+        local score = ARGV[2]
+        local serializedItem = ARGV[3]
+        local deduplicationKey = ARGV[4]
+
+        -- Leave the item alone if, while this attempt ran, it was removed (acked or cancelled)
+        -- or a newer enqueue replaced it.
+        if deduplicationKey ~= "" then
+          local current = redis.call('HGET', items, id)
+          if not current then
+            return 2
+          end
+          local success, parsed = pcall(cjson.decode, current)
+          if success and parsed.deduplicationKey and parsed.deduplicationKey ~= deduplicationKey then
+            return 2
+          end
+        end
+
+        redis.call('ZADD', queue, score, id)
+        redis.call('HSET', items, id, serializedItem)
+
+        return 1
+      `,
+    });
+
     this.redis.defineCommand("moveToDeadLetterQueue", {
       numberOfKeys: 4,
       lua: `
@@ -605,13 +690,17 @@ export class SimpleQueue<TMessageCatalog extends MessageCatalogSchema> {
         local dlqItems = KEYS[4]
         local id = ARGV[1]
         local errorMessage = ARGV[2]
+        local deduplicationKey = ARGV[3]
 
         local item = redis.call('HGET', items, id)
         if not item then
           return 0
         end
 
         local parsedItem = cjson.decode(item)
+        if deduplicationKey ~= "" and parsedItem.deduplicationKey and parsedItem.deduplicationKey ~= deduplicationKey then
+          return 2
+        end
         parsedItem.errorMessage = errorMessage
 
         local time = redis.call('TIME')
@@ -738,6 +827,17 @@ declare module "@internal/redis" {
       dlqItems: string,
       id: string,
       errorMessage: string,
+      deduplicationKey: string,
+      callback?: Callback<number>
+    ): Result<number, Context>;
+
+    requeueItem(
+      queue: string,
+      items: string,
+      id: string,
+      score: number,
+      serializedItem: string,
+      deduplicationKey: string,
       callback?: Callback<number>
     ): Result<number, Context>;
 

packages/redis-worker/src/worker.test.ts

@@ -694,4 +694,60 @@ describe("Worker", () => {
       await worker.stop();
     }
   );
+
+  redisTest(
+    "A failed attempt does not overwrite or dead-letter a newer enqueue of the same id",
+    { timeout: 30_000 },
+    async ({ redisContainer }) => {
+      for (const maxAttempts of [1, 3]) {
+        const seen: Array<{ value: number; attempt: number }> = [];
+        let release!: () => void;
+        const released = new Promise<void>((resolve) => (release = resolve));
+        let started!: () => void;
+        const firstStarted = new Promise<void>((resolve) => (started = resolve));
+
+        const worker = new Worker({
+          name: `test-supersede-${maxAttempts}`,
+          redisOptions: {
+            host: redisContainer.getHost(),
+            port: redisContainer.getPort(),
+            password: redisContainer.getPassword(),
+          },
+          catalog: {
+            testJob: {
+              schema: z.object({ value: z.number() }),
+              visibilityTimeoutMs: 5000,
+              retry: { maxAttempts, minTimeoutInMs: 10_000, maxTimeoutInMs: 10_000 },
+            },
+          },
+          jobs: {
+            testJob: async ({ payload, attempt }) => {
+              seen.push({ value: payload.value, attempt });
+              if (payload.value !== 1) return;
+              started();
+              await released;
+              throw new Error("first attempt failed");
+            },
+          },
+          concurrency: { workers: 1, tasksPerWorker: 2 },
+          pollIntervalMs: 20,
+          logger: new Logger("test", "error"),
+        }).start();
+
+        await worker.enqueue({ id: "same", job: "testJob", payload: { value: 1 } });
+        await firstStarted;
+        await worker.enqueue({ id: "same", job: "testJob", payload: { value: 2 } });
+        release();
+
+        await new Promise((resolve) => setTimeout(resolve, 1000));
+        expect(await worker.queue.sizeOfDeadLetterQueue()).toBe(0);
+        await worker.stop();
+
+        // The newer enqueue ran on its own first attempt, and the failed attempt neither
+        // requeued its stale payload over it nor moved it to the dead letter queue.
+        expect(seen).toContainEqual({ value: 2, attempt: 0 });
+        expect(seen.filter((s) => s.value === 1)).toHaveLength(1);
+      }
+    }
+  );
 });