fix(flow-producer): surface aborted transactions instead of silent fail

src/classes/flow-producer.ts

@@ -225,14 +225,28 @@ export class FlowProducer extends EventEmitter {
         const results = (await multi.exec()) as
           | [null | Error, string | number][]
           | null;
-        const [result] = results || [];
-        if (result) {
-          const [err, jobId] = result;
-          if (!err && typeof jobId === 'string') {
-            jobsTree.job.id = jobId;
+
+        // Surface transaction-level and per-command failures instead
+        // of silently returning a half-formed JobNode (issue #3851).
+        // ioredis returns `null` from exec() when the multi was
+        // aborted (READONLY replica, WATCH conflict, pipeline error)
+        // and `[err, value]` per command otherwise.
+        if (results === null) {
+          throw new Error(
+            'Flow could not be added: Redis transaction was aborted',
+          );
+        }
+        for (const [err] of results) {
+          if (err) {
+            throw err;
           }
         }
 
+        const [, jobId] = results[0] ?? [];
+        if (typeof jobId === 'string') {
+          jobsTree.job.id = jobId;
+        }
+
         return jobsTree;
       },
     );
@@ -302,8 +316,22 @@ export class FlowProducer extends EventEmitter {
         const results = (await multi.exec()) as
           | [null | Error, string | number][]
           | null;
+
+        // Surface a transaction-level abort (e.g. READONLY replica or
+        // pipeline error) so callers can distinguish "nothing was
+        // added" from "some flows partially failed" (issue #3851).
+        // Per-command errors are NOT thrown here on purpose: callers
+        // can detect partially-failed roots via the returned tree's
+        // missing job id (and there is an existing test that relies
+        // on that semantics for addBulk).
+        if (results === null) {
+          throw new Error(
+            'Flows could not be added: Redis transaction was aborted',
+          );
+        }
+
         for (let index = 0; index < jobsTrees.length; ++index) {
-          const result = results?.[index];
+          const result = results[index];
           if (!result) {
             continue;
           }

tests/flow.test.ts

@@ -7,6 +7,7 @@ import {
   afterAll,
   it,
   expect,
+  vi,
 } from 'vitest';
 
 import {
@@ -6322,6 +6323,62 @@ describe('flows', () => {
       await flow.close();
     });
 
+    // Regression for https://github.com/taskforcesh/bullmq/issues/3851:
+    // FlowProducer.add() and addBulk() ignored a `null` return from
+    // `multi.exec()` (which ioredis emits when the transaction is
+    // aborted, e.g. against a READONLY replica). The reporter saw
+    // .add() resolve with a half-formed JobNode whose id pointed at
+    // nothing in Redis. The producer must surface the abort.
+    it('throws when redis transaction is aborted on FlowProducer.add', async () => {
+      const flow = new FlowProducer({ connection, prefix });
+      await flow.waitUntilReady();
+
+      const client = (await flow.client) as IORedis;
+      const realMulti = client.multi.bind(client);
+      const multiSpy = vi
+        .spyOn(client, 'multi')
+        .mockImplementationOnce((...args: any[]) => {
+          const m = realMulti(...args);
+          // Override exec() to mirror the ioredis behaviour when the
+          // transaction is aborted: a single null return.
+          (m as any).exec = async () => null;
+          return m;
+        });
+
+      try {
+        await expect(
+          flow.add({ name: 'parent', data: {}, queueName }),
+        ).rejects.toThrow(/transaction was aborted/);
+      } finally {
+        multiSpy.mockRestore();
+        await flow.close();
+      }
+    });
+
+    it('throws when redis transaction is aborted on FlowProducer.addBulk', async () => {
+      const flow = new FlowProducer({ connection, prefix });
+      await flow.waitUntilReady();
+
+      const client = (await flow.client) as IORedis;
+      const realMulti = client.multi.bind(client);
+      const multiSpy = vi
+        .spyOn(client, 'multi')
+        .mockImplementationOnce((...args: any[]) => {
+          const m = realMulti(...args);
+          (m as any).exec = async () => null;
+          return m;
+        });
+
+      try {
+        await expect(
+          flow.addBulk([{ name: 'parent', data: {}, queueName }]),
+        ).rejects.toThrow(/transaction was aborted/);
+      } finally {
+        multiSpy.mockRestore();
+        await flow.close();
+      }
+    });
+
     it('should not corrupt id mapping for successful jobs when some addBulk commands fail', async () => {
       const flow = new FlowProducer({ connection, prefix });