Fix subscribed_to maybe empty

celery/backends/redis.py

@@ -103,17 +103,25 @@ def _reconnect_pubsub(self):
         self.backend.client.connection_pool.reset()
         # task state might have changed when the connection was down so we
         # retrieve meta for all subscribed tasks before going into pubsub mode
-        metas = self.backend.client.mget(self.subscribed_to)
-        metas = [meta for meta in metas if meta]
-        for meta in metas:
-            self.on_state_change(self._decode_result(meta), None)
+        if self.subscribed_to:
+            metas = self.backend.client.mget(self.subscribed_to)
+            metas = [meta for meta in metas if meta]
+            for meta in metas:
+                self.on_state_change(self._decode_result(meta), None)
         self._pubsub = self.backend.client.pubsub(
             ignore_subscribe_messages=True,
         )
+        # subscribed_to maybe empty after on_state_change
         if self.subscribed_to:
             self._pubsub.subscribe(*self.subscribed_to)
         else:
-            self._pubsub.ping()
+            self._pubsub.connection = self._pubsub.connection_pool.get_connection(
+                'pubsub', self._pubsub.shard_hint
+            )
+            # even if nothing to subscribe, we should not lost the callback after being connected.
+            # maybe on_connect will be changed in the future. at present, this callback
+            # re-subscribe to any channels previously subscribed to.
+            self._pubsub.connection.register_connect_callback(self._pubsub.on_connect)
 
     @contextmanager
     def reconnect_on_error(self):

t/unit/backends/test_redis.py

@@ -286,6 +286,35 @@ def test_drain_events_connection_error_no_patch(self):
         consumer.drain_events()
         consumer._pubsub.subscribe.assert_not_called()
 
+    def test__reconnect_pubsub_no_subscribed(self):
+        consumer = self.get_consumer()
+        consumer.start('initial')
+        consumer.subscribed_to = set()
+        consumer._reconnect_pubsub()
+        consumer.backend.client.mget.assert_not_called()
+        consumer._pubsub.subscribe.assert_not_called()
+        consumer._pubsub.connection.register_connect_callback.assert_called_once()
+
+    def test__reconnect_pubsub_with_state_change(self):
+        meta = {'task_id': 'initial', 'status': states.SUCCESS}
+        consumer = self.get_consumer()
+        consumer.start('initial')
+        consumer.backend._set_with_state(b'celery-task-meta-initial', json.dumps(meta), states.SUCCESS)
+        consumer._reconnect_pubsub()
+        consumer.backend.client.mget.assert_called_once()
+        consumer._pubsub.subscribe.assert_not_called()
+        consumer._pubsub.connection.register_connect_callback.assert_called_once()
+
+    def test__reconnect_pubsub_without_state_change(self):
+        meta = {'task_id': 'initial', 'status': states.STARTED}
+        consumer = self.get_consumer()
+        consumer.start('initial')
+        consumer.backend._set_with_state(b'celery-task-meta-initial', json.dumps(meta), states.SUCCESS)
+        consumer._reconnect_pubsub()
+        consumer.backend.client.mget.assert_called_once()
+        consumer._pubsub.subscribe.assert_called_once()
+        consumer._pubsub.connection.register_connect_callback.assert_not_called()
+
 
 class basetest_RedisBackend:
     def get_backend(self):