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

[mohanrajvenkatesan23-04]

Port Impact Checklist

• Python – does this change need to be ported or documented in the Python library?
• Elixir – does this change need to be ported or documented in the Elixir library?
• PHP – does this change need to be ported or documented in the PHP library?

Why

Fixes #3851.

FlowProducer.add() and FlowProducer.addBulk() swallowed multi.exec() returning null. ioredis returns null from Pipeline.exec() whenever the transaction was aborted — for example issued against a READONLY replica, a WATCH conflict, or a pipeline-level error — and the producer would happily resolve with a half-formed JobNode whose job.id was the locally-generated UUID, pointing at nothing in Redis.

The reporter saw exactly that shape under an Upstash failover window where Redis is briefly read-only:

Adding jobs with FlowProducer...
Job added with FlowProducer unexpectedly: 6d82aace-3808-4750-8040-f9feb5149023
.add() did not throw but flow does not exist: undefined

Queue.add() correctly throws in the same scenario; the bug was in flow-producer.ts:225 and the parallel block in addBulk at line 302.

How

Two paths needed slightly different treatment because the API contract differs.

• add() (single flow — atomic by design): throw when results === null, and throw the first per-command error if any. Without this, a transaction abort or a per-command failure leaves the user with a JobNode whose UUID id is unrecoverable.
• addBulk() (partial-success semantics): an existing regression test (should not corrupt id mapping for successful jobs when some addBulk commands fail) explicitly relies on addBulk letting some root commands fail while others succeed. So we throw only on results === null (the entire transaction aborted, every flow is dead). Per-command errors continue to leave that root's job.id unassigned, and callers detect partial failure via await queue.getJob(tree.job.id) === undefined — the same contract that test asserts.

The error message names the cause:

Flow could not be added: Redis transaction was aborted

Additional Notes (Optional)

• Two new regression tests in tests/flow.test.ts use a vitest spy on the next client.multi() call to swap in a multi whose .exec() returns null. Both add() and addBulk() are asserted to throw /transaction was aborted/. The original Pipeline is otherwise pristine — no impact on neighbouring tests.
• The existing partial-success test (should not corrupt id mapping...) is preserved unchanged and still passes against the new logic.
• TypeScript / Node-only. No port work required.

[manast]

@copilot resolve the merge conflicts in this pull request

[Copilot]

Pull request overview

This PR fixes a FlowProducer edge case where FlowProducer.add() / addBulk() could silently succeed when multi.exec() returns null (aborted transaction), returning a JobNode whose ID doesn’t correspond to anything in Redis.

Changes:

• FlowProducer.add(): throws when multi.exec() returns null and now surfaces per-command Errors from exec() results.
• FlowProducer.addBulk(): throws when multi.exec() returns null while preserving existing partial-success semantics for per-command failures.
• Adds regression tests that mock multi.exec() to return null and assert both APIs throw.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 4 comments.

File	Description
src/classes/flow-producer.ts	Adds explicit handling for aborted transactions (exec() === null) and improves error surfacing behavior.
tests/flow.test.ts	Adds regression coverage for aborted transactions via a mocked multi.exec() returning null.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[manast]

@copilot resolve the merge conflicts in this pull request

[mohanrajvenkatesan23-04]

Thanks for the review! Pushed a follow-up commit addressing all four comments:

• flow-producer.ts (add and addBulk): switched the aborted-transaction guard to an explicit results === null check so an empty exec() result is no longer misclassified as an abort.
• flow.test.ts (both spots): now use the public flow.client getter instead of (flow as any).connection.client, dropping the any cast and the coupling to internals.