fix(flow-producer): reject deduplication and debounce options (#2780)

src/classes/flow-producer.ts

@@ -1,5 +1,6 @@
 import { EventEmitter } from 'events';
 import {
+  FlowChildJob,
   FlowJob,
   FlowQueuesOpts,
   FlowOpts,
@@ -210,6 +211,7 @@ export class FlowProducer extends EventEmitter {
     if (this.closing) {
       return;
     }
+    assertNoDeduplication(flow);
     const client = await this.connection.client;
     const multi = client.multi();
 
@@ -304,6 +306,7 @@ export class FlowProducer extends EventEmitter {
     if (this.closing) {
       return;
     }
+    flows.forEach(assertNoDeduplication);
     const client = await this.connection.client;
     const multi = client.multi();
 
@@ -675,3 +678,25 @@ export class FlowProducer extends EventEmitter {
     return this.connection.disconnect();
   }
 }
+
+// Deduplication is incompatible with FlowProducer because parent ids must be
+// known before the multi.exec() commits. A dedup hit returns the existing
+// job's id rather than the freshly generated one, which would orphan the
+// flow's children. See https://github.com/taskforcesh/bullmq/issues/2780.
+//
+// The whole tree is walked, not just the root: FlowChildJob omits these
+// options at the type level, but JavaScript callers (and TypeScript escape
+// hatches) can still set them on a child, and addChildren() would pass them
+// straight through to Job.addJob.
+function assertNoDeduplication(flow: FlowJob | FlowChildJob): void {
+  const opts = flow?.opts as FlowJob['opts'];
+
+  if (opts?.deduplication || opts?.debounce) {
+    throw new Error(
+      `Deduplication and debounce are not supported in flows ` +
+        `(queue "${flow.queueName}", job "${flow.name}").`,
+    );
+  }
+
+  flow?.children?.forEach(child => assertNoDeduplication(child));
+}

tests/flow.test.ts

@@ -6550,4 +6550,115 @@ describe('flows', () => {
       await flow.close();
     });
   });
+
+  describe('when deduplication or debounce is provided to a flow', () => {
+    it('rejects add() with a clear error', async () => {
+      const flow = new FlowProducer({ connection, prefix });
+
+      await expect(
+        flow.add({
+          name: 'parent',
+          data: {},
+          queueName,
+          opts: { deduplication: { id: 'dedup-id' } },
+          children: [{ name: 'child', data: {}, queueName }],
+        }),
+      ).rejects.toThrow(/Deduplication and debounce are not supported in flows/);
+
+      await expect(
+        flow.add({
+          name: 'parent',
+          data: {},
+          queueName,
+          opts: { debounce: { id: 'debounce-id' } },
+          children: [{ name: 'child', data: {}, queueName }],
+        }),
+      ).rejects.toThrow(/Deduplication and debounce are not supported in flows/);
+
+      await flow.close();
+    });
+
+    it('rejects add() when a nested child uses deduplication or debounce', async () => {
+      const flow = new FlowProducer({ connection, prefix });
+
+      // FlowChildJob omits these options, so a cast is needed to reproduce
+      // what a JavaScript caller can pass at runtime.
+      await expect(
+        flow.add({
+          name: 'parent',
+          data: {},
+          queueName,
+          children: [
+            {
+              name: 'child',
+              data: {},
+              queueName,
+              children: [
+                {
+                  name: 'grandchild',
+                  data: {},
+                  queueName,
+                  opts: { deduplication: { id: 'dedup-id' } },
+                } as any,
+              ],
+            },
+          ],
+        }),
+      ).rejects.toThrow(/Deduplication and debounce are not supported in flows/);
+
+      await expect(
+        flow.add({
+          name: 'parent',
+          data: {},
+          queueName,
+          children: [
+            {
+              name: 'child',
+              data: {},
+              queueName,
+              opts: { debounce: { id: 'debounce-id' } },
+            } as any,
+          ],
+        }),
+      ).rejects.toThrow(/Deduplication and debounce are not supported in flows/);
+
+      await flow.close();
+    });
+
+    it('rejects addBulk() if any flow uses deduplication', async () => {
+      const flow = new FlowProducer({ connection, prefix });
+
+      await expect(
+        flow.addBulk([
+          { name: 'first', data: {}, queueName, opts: {} },
+          {
+            name: 'second',
+            data: {},
+            queueName,
+            opts: { deduplication: { id: 'dedup-id' } },
+          },
+        ]),
+      ).rejects.toThrow(/Deduplication and debounce are not supported in flows/);
+
+      await flow.close();
+    });
+
+    it('rejects addBulk() if any flow uses debounce', async () => {
+      const flow = new FlowProducer({ connection, prefix });
+
+      await expect(
+        flow.addBulk([
+          { name: 'first', data: {}, queueName, opts: {} },
+          {
+            name: 'second',
+            data: {},
+            queueName,
+            opts: { debounce: { id: 'debounce-id' } },
+          },
+        ]),
+      ).rejects.toThrow(/Deduplication and debounce are not supported in flows/);
+
+      await flow.close();
+    });
+  });
 });