fix(pubsub): avoid UnicodeDecodeError on reconnect with binary channel names

redis/asyncio/client.py

@@ -959,18 +959,38 @@ async def on_connect(self, connection: Connection):
         # NOTE: for python3, we can't pass bytestrings as keyword arguments
         # so we need to decode channel/pattern names back to unicode strings
         # before passing them to [p]subscribe.
+        #
+        # However, channels subscribed without a callback (positional args) may
+        # have binary names that are not valid in the current encoding (e.g.
+        # arbitrary bytes that are not valid UTF-8).  These channels are stored
+        # with a ``None`` handler.  We re-subscribe them as positional args so
+        # that no decoding is required.
         self.pending_unsubscribe_channels.clear()
         self.pending_unsubscribe_patterns.clear()
         if self.channels:
-            channels = {}
+            channels_with_handlers = {}
+            channels_without_handlers = []
             for k, v in self.channels.items():
-                channels[self.encoder.decode(k, force=True)] = v
-            await self.subscribe(**channels)
+                if v is not None:
+                    channels_with_handlers[self.encoder.decode(k, force=True)] = v
+                else:
+                    channels_without_handlers.append(k)
+            if channels_with_handlers or channels_without_handlers:
+                await self.subscribe(
+                    *channels_without_handlers, **channels_with_handlers
+                )
         if self.patterns:
-            patterns = {}
+            patterns_with_handlers = {}
+            patterns_without_handlers = []
             for k, v in self.patterns.items():
-                patterns[self.encoder.decode(k, force=True)] = v
-            await self.psubscribe(**patterns)
+                if v is not None:
+                    patterns_with_handlers[self.encoder.decode(k, force=True)] = v
+                else:
+                    patterns_without_handlers.append(k)
+            if patterns_with_handlers or patterns_without_handlers:
+                await self.psubscribe(
+                    *patterns_without_handlers, **patterns_with_handlers
+                )
 
     @property
     def subscribed(self):

redis/client.py

@@ -898,25 +898,45 @@ def on_connect(self, connection) -> None:
         # NOTE: for python3, we can't pass bytestrings as keyword arguments
         # so we need to decode channel/pattern names back to unicode strings
         # before passing them to [p]subscribe.
+        #
+        # However, channels subscribed without a callback (positional args) may
+        # have binary names that are not valid in the current encoding (e.g.
+        # arbitrary bytes that are not valid UTF-8).  These channels are stored
+        # with a ``None`` handler.  We re-subscribe them as positional args so
+        # that no decoding is required.
         self.pending_unsubscribe_channels.clear()
         self.pending_unsubscribe_patterns.clear()
         self.pending_unsubscribe_shard_channels.clear()
         if self.channels:
-            channels = {
-                self.encoder.decode(k, force=True): v for k, v in self.channels.items()
-            }
-            self.subscribe(**channels)
+            channels_with_handlers = {}
+            channels_without_handlers = []
+            for k, v in self.channels.items():
+                if v is not None:
+                    channels_with_handlers[self.encoder.decode(k, force=True)] = v
+                else:
+                    channels_without_handlers.append(k)
+            if channels_with_handlers or channels_without_handlers:
+                self.subscribe(*channels_without_handlers, **channels_with_handlers)
         if self.patterns:
-            patterns = {
-                self.encoder.decode(k, force=True): v for k, v in self.patterns.items()
-            }
-            self.psubscribe(**patterns)
+            patterns_with_handlers = {}
+            patterns_without_handlers = []
+            for k, v in self.patterns.items():
+                if v is not None:
+                    patterns_with_handlers[self.encoder.decode(k, force=True)] = v
+                else:
+                    patterns_without_handlers.append(k)
+            if patterns_with_handlers or patterns_without_handlers:
+                self.psubscribe(*patterns_without_handlers, **patterns_with_handlers)
         if self.shard_channels:
-            shard_channels = {
-                self.encoder.decode(k, force=True): v
-                for k, v in self.shard_channels.items()
-            }
-            self.ssubscribe(**shard_channels)
+            shard_with_handlers = {}
+            shard_without_handlers = []
+            for k, v in self.shard_channels.items():
+                if v is not None:
+                    shard_with_handlers[self.encoder.decode(k, force=True)] = v
+                else:
+                    shard_without_handlers.append(k)
+            if shard_with_handlers or shard_without_handlers:
+                self.ssubscribe(*shard_without_handlers, **shard_with_handlers)
 
     @property
     def subscribed(self) -> bool:

tests/test_asyncio/test_pubsub.py

@@ -160,6 +160,50 @@ async def test_resubscribe_to_patterns_on_reconnection(self, pubsub):
         kwargs = make_subscribe_test_data(pubsub, "pattern")
         await self._test_resubscribe_on_reconnection(**kwargs)
 
+    async def test_resubscribe_binary_channel_on_reconnection(self, pubsub):
+        """Binary channel names that are not valid UTF-8 must survive
+        reconnection without raising ``UnicodeDecodeError``.
+        See https://github.com/redis/redis-py/issues/3912
+        """
+        # b'\x80\x81\x82' is deliberately invalid UTF-8
+        binary_channel = b"\x80\x81\x82"
+        p = pubsub
+        await p.subscribe(binary_channel)
+        assert await wait_for_message(p) is not None  # consume subscribe ack
+
+        # force reconnect
+        await p.connection.disconnect()
+
+        # get_message triggers on_connect → re-subscribe; must not raise
+        messages = []
+        for _ in range(1):
+            messages.append(await wait_for_message(p))
+
+        assert len(messages) == 1
+        assert messages[0]["type"] == "subscribe"
+        assert messages[0]["channel"] == binary_channel
+
+    async def test_resubscribe_binary_pattern_on_reconnection(self, pubsub):
+        """Binary pattern names that are not valid UTF-8 must survive
+        reconnection without raising ``UnicodeDecodeError``.
+        See https://github.com/redis/redis-py/issues/3912
+        """
+        binary_pattern = b"\x80\x81*"
+        p = pubsub
+        await p.psubscribe(binary_pattern)
+        assert await wait_for_message(p) is not None  # consume psubscribe ack
+
+        # force reconnect
+        await p.connection.disconnect()
+
+        messages = []
+        for _ in range(1):
+            messages.append(await wait_for_message(p))
+
+        assert len(messages) == 1
+        assert messages[0]["type"] == "psubscribe"
+        assert messages[0]["channel"] == binary_pattern
+
     async def _test_subscribed_property(
         self, p, sub_type, unsub_type, sub_func, unsub_func, keys
     ):

tests/test_pubsub.py

@@ -196,6 +196,52 @@ def test_resubscribe_to_shard_channels_on_reconnection(self, r):
         kwargs = make_subscribe_test_data(r.pubsub(), "shard_channel")
         self._test_resubscribe_on_reconnection(**kwargs)
 
+    @pytest.mark.onlynoncluster
+    def test_resubscribe_binary_channel_on_reconnection(self, r):
+        """Binary channel names that are not valid UTF-8 must survive
+        reconnection without raising ``UnicodeDecodeError``.
+        See https://github.com/redis/redis-py/issues/3912
+        """
+        # b'\x80\x81\x82' is deliberately invalid UTF-8
+        binary_channel = b"\x80\x81\x82"
+        p = r.pubsub()
+        p.subscribe(binary_channel)
+        assert wait_for_message(p) is not None  # consume subscribe ack
+
+        # force reconnect
+        p.connection.disconnect()
+
+        # get_message triggers on_connect → re-subscribe; must not raise
+        messages = []
+        for _ in range(1):
+            messages.append(wait_for_message(p))
+
+        assert len(messages) == 1
+        assert messages[0]["type"] == "subscribe"
+        assert messages[0]["channel"] == binary_channel
+
+    @pytest.mark.onlynoncluster
+    def test_resubscribe_binary_pattern_on_reconnection(self, r):
+        """Binary pattern names that are not valid UTF-8 must survive
+        reconnection without raising ``UnicodeDecodeError``.
+        See https://github.com/redis/redis-py/issues/3912
+        """
+        binary_pattern = b"\x80\x81*"
+        p = r.pubsub()
+        p.psubscribe(binary_pattern)
+        assert wait_for_message(p) is not None  # consume psubscribe ack
+
+        # force reconnect
+        p.connection.disconnect()
+
+        messages = []
+        for _ in range(1):
+            messages.append(wait_for_message(p))
+
+        assert len(messages) == 1
+        assert messages[0]["type"] == "psubscribe"
+        assert messages[0]["channel"] == binary_pattern
+
     def _test_subscribed_property(
         self, p, sub_type, unsub_type, sub_func, unsub_func, keys
     ):
@@ -1122,9 +1168,9 @@ def test_get_message_wait_for_subscription_not_being_called(self, r):
         # Ensure p has the event attribute your wait_for_message would call:
         ev = getattr(p, "subscribed_event", None)
 
-        assert ev is not None, (
-            "PubSub event attribute not found (check redis-py version)"
-        )
+        assert (
+            ev is not None
+        ), "PubSub event attribute not found (check redis-py version)"
 
         with patch.object(ev, "wait") as mock:
             assert wait_for_message(p) == make_message("subscribe", "foo", 1)