fix(worker): [python] isolate blocking client when sharing Redis connection

[mohanrajvenkatesan23-04]

Fixes #3401

Why

When a user passes the same redis.asyncio.Redis instance to both Queue and Worker, job.data arrives as an empty dict and job.name as None inside the processor. Using Redis URL strings works fine.

Since #3887 unified Python connections on single_connection_client=True, Worker.__init__ calls RedisConnection(redis_opts) twice — once for the regular client and once for the blocking client. When redis_opts is an already-constructed redis.Redis, RedisConnection reused that same instance verbatim for both connections, so the Worker's BZPOPMIN blocking command monopolised the single underlying socket and interleaved with / corrupted replies of subsequent commands (the HMGET that reads the job hash after moveToActive). Node avoids this by calling .duplicate() on externally-supplied clients (src/classes/worker.ts) and creating the blocking RedisConnection with shared: false (src/classes/redis-connection.ts).

How

• RedisConnection now tracks a shared flag whenever the caller supplies an already-constructed Redis instance; close() and disconnect() are no-ops in that case so the caller retains ownership of their client.
• RedisConnection gained an isBlocking flag. When isBlocking=True and the caller supplied a Redis instance, the constructor derives a sibling client via Redis.client() so blocking operations run on a dedicated connection from the same pool.
• Worker constructs its blockingRedisConnection with isBlocking=True so BZPOPMIN can no longer starve the regular client's replies.

Additional Notes (Optional)

• New python/tests/shared_connection_test.py — 6 mock-based regression tests covering the shared flag, close/disconnect being no-ops, reuse of the caller's client on the non-blocking path, duplication on the blocking path, and an end-to-end assertion that worker.client is not worker.bclient when a shared client is passed.
• flake8 --select=E9,F63,F7,F82 bullmq tests is clean; existing non-integration tests still pass.
• No Node / TypeScript source was touched; commit tagged [python].

[Copilot]

Pull request overview

Fixes a Python-only bug where sharing the same redis.asyncio.Redis instance between Queue and Worker can corrupt replies when the worker issues blocking commands, leading to empty job.data / None job.name.

Changes:

• Add shared + isBlocking behavior to RedisConnection to treat externally-supplied clients as caller-owned and to derive a dedicated blocking client.
• Update Worker to create its blocking connection with isBlocking=True.
• Add a regression test suite covering shared-client behavior and blocking-client isolation.

Reviewed changes

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

File	Description
python/bullmq/redis_connection.py	Track externally-owned clients and attempt to isolate blocking usage via a derived sibling client.
python/bullmq/worker.py	Use isBlocking=True for the worker’s blocking Redis connection.
python/tests/shared_connection_test.py	Add regression tests for shared client + blocking-client isolation behavior.

---

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

[mohanrajvenkatesan23-04]

Thanks for the review @manast! Pushed a follow-up commit addressing every point:

• Long inline comments trimmed in redis_connection.py (the block at line 79 and the close() docstring at line 135) and worker.py (line 42).
• close() docstring reworded to describe behaviour rather than prescribe action — now reads "A shared connection is left open; it remains owned by the caller."
• Test file no longer issue-specific — class renamed to TestSharedConnection, the docstring describes shared-connection behaviour generally, and the issue references are gone. The file is named shared_connection_test.py so it can serve as the home for any future shared-connection tests; happy to fold it into another existing suite if you'd prefer one.
• Ownership fix for Copilot's concern — only the directly-reused caller client is now flagged as shared. The derived blocking sibling has shared=False, so worker.close() releases it cleanly without touching the caller-owned client. Added an end-to-end assertion that confirms worker.bclient.aclose() runs while caller.aclose is left alone.

All seven shared-connection tests pass locally. Could you take another look when you have a moment? Happy to iterate on anything that still feels off.

[Copilot]

Pull request overview

Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.

---

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

[mohanrajvenkatesan23-04]

Thanks for the suggestion! Pushed a follow-up commit (0c1afcd) adding a real-Redis integration test in python/tests/shared_connection_test.py:

• New TestSharedConnectionIntegration class (uses unittest.IsolatedAsyncioTestCase).
• Builds a single redis.asyncio.Redis client and hands it to both Queue and Worker via opts["connection"].
• Adds a job with a non-trivial name and nested data, awaits the worker's processor (10s timeout), and asserts both job.name and job.data are intact — directly exercising the original interleaved-reply symptom from [Bug]: Shared Redis connection objects cause job data to become empty and job names to become None #3401.
• Also asserts the caller's shared client is still pingable after worker.close() (i.e. the worker only released its derived blocking sibling).
• flushdb is run on asyncSetUp/asyncTearDown so the test is hermetic alongside the existing CI Redis services.

The 7 mock-based unit tests continue to pass locally; the new integration test will run against the redis@7-alpine, valkey@8, and dragonflydb@latest services in CI.