run_in_thread dies if reconnection fails

[jkarneges]

PubSubWorkerThread.run loops on a call to get_message. If the connection is lost, this call will silently attempt reconnection. However, if a single reconnect attempt fails, then an exception will bubble up. Since there is nobody to catch the exception, the thread dies.

An easy way to reproduce the issue is to set up a subscription and then restart the Redis server. On my machine, Redis takes a few seconds to restart. redis-py will attempt to reconnect faster than this and get a connection refused error. I would have expected redis-py to simply eventually reconnect after the Redis server comes back up.

I believe what is needed here is a way for the PubSubWorkerThread to reconnect multiple times until success. Putting a try/except around the call to get_message does the trick, but I don't know if that's the right answer.

[jkarneges]

In the meantime, for anyone looking for a workaround, I'm using a custom thread class:

class MyPubSubWorkerThread(threading.Thread):
    def __init__(self, pubsub, sleep_time, daemon=False):
        super(MyPubSubWorkerThread, self).__init__()
        self.daemon = daemon
        self.pubsub = pubsub
        self.sleep_time = sleep_time
        self._running = False

    def run(self):
        if self._running:
            return
        self._running = True
        pubsub = self.pubsub
        sleep_time = self.sleep_time
        while pubsub.subscribed:
            try:
                pubsub.get_message(ignore_subscribe_messages=True,
                            timeout=sleep_time)
            except:
                logger.exception('get_message failed')
                time.sleep(2)
        pubsub.close()
        self._running = False

    def stop(self):
        # stopping simply unsubscribes from all channels and patterns.
        # the unsubscribe responses that are generated will short circuit
        # the loop in run(), calling pubsub.close() to clean up the connection
        self.pubsub.unsubscribe()
        self.pubsub.punsubscribe()

This is basically a copy of redis-py's PubSubWorkerThread, but with a try/except around the call to get_message. Also there's a sleep after an exception so it doesn't hammer the server.

Use it like this:

p = r.pubsub()
p.subscribe(...)
thread = MyPubSubWorkerThread(p, sleep_time=1, daemon=True)
thread.start()

[AlexEshoo]

The above implementation is a decent way to ride through connection loss and simply keep retying until connection is re-established, but a few changes and features may make it better to incorporate into the module.

It would be nice if there could be a registered callback on the exception so that if you just want to shut down and do something else fancy, you can.

The exceptions to catch should also be specified by the user (not just except on everything) This way callbacks can be configured on a per exception basis. Similar to how tenacity does it perhaps?

The time to sleep for should also be a parameter I think.

[github-actions]

This issue is marked stale. It will be closed in 30 days if it is not updated.