PubSubBroker listener close leaks subscriptions and exhausts its connection pool #143

Closing PubSubBroker.listen() leaves its Redis subscription and its pool connection checked out. With a bounded pool, the next publish or replacement listener then fails with ConnectionError: No connection available.

The listener owns a PubSub object, but only the surrounding Redis client is context-managed. That client uses the broker's externally owned pool, so leaving its context does not close the subscription or release its dedicated connection.

Reproduction

Run against a local Redis server. TEST_REDIS_URL can select another endpoint (the reproduction here used a private local instance and ?protocol=2 for Redis 5).

import asyncio
import os

from redis.asyncio import Redis
from taskiq import BrokerMessage
from taskiq_redis import PubSubBroker


async def main():
    url = os.environ.get('TEST_REDIS_URL', 'redis://127.0.0.1:6379')
    broker = PubSubBroker(url, queue_name='cleanup-reproduction', max_connection_pool_size=1, timeout=0.1)
    async with Redis.from_url(url) as observer:
        listener = broker.listen()
        read = asyncio.create_task(anext(listener))
        try:
            async with asyncio.timeout(3):
                while (await observer.pubsub_numsub(broker.queue_name))[0][1] != 1:
                    await asyncio.sleep(0)
            await observer.publish(broker.queue_name, b'first')
            print('first payload:', await read)
            await listener.aclose()
            print('subscribers after aclose:', (await observer.pubsub_numsub(broker.queue_name))[0][1])
            try:
                await broker.kick(BrokerMessage(task_id='next', task_name='next', message=b'next', labels={}))
                print('publishing with the same pool: succeeded')
            except Exception as exc:
                print('publishing with the same pool:', type(exc).__name__, str(exc))
        finally:
            read.cancel()
            await asyncio.gather(read, return_exceptions=True)
            await listener.aclose()
            await broker.shutdown()


asyncio.run(main())

Actual output on unchanged main:

first payload: b'first'
subscribers after aclose: 1
publishing with the same pool: ConnectionError No connection available.

Expected: closing the listener unsubscribes/disconnects its PubSub connection and releases the slot, so publishing and opening another listener can proceed. The same cleanup should happen when a pending receive is cancelled or fails.

Environment and proposed scope

Main at 8f70bc85dd6a469cae7d3b467af7dea720c92434, taskiq 0.13.0, redis-py 8.0.0, Redis server 5.0.14.1 over actual loopback TCP/RESP2, Python 3.14.7, Windows 11. No fake Redis or patched internals were used.

I would like to context-manage the PubSub object inside PubSubBroker.listen() and add real-Redis regressions for generator close, cancellation, failure cleanup, and reuse. This does not change Redis Streams, delivery guarantees, or the public API. I checked the open PR diffs (#129, #141, #142); their shared-file changes concern other broker implementations. Could a maintainer assign this issue to cw-jlu?

Investigation, reproduction, and this report used OpenAI Codex assistance.