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

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

Channels subscribed as positional arguments (without a callback handler)
may carry binary names that are not valid in the connection's encoding
(e.g. arbitrary bytes that are not valid UTF-8).  The existing
`on_connect` method decoded every channel name via `force=True` to pass
them as keyword arguments to `subscribe`/`psubscribe`, which raised
`UnicodeDecodeError` for these channels.

Split the reconnection logic: channels with handlers are decoded and
passed as kwargs (they were originally subscribed as kwargs, so their
names are guaranteed decodable); channels without handlers are passed as
positional args, preserving the original bytes.

Applied the same fix to:
- async PubSub.on_connect (channels and patterns)
- sync PubSub.on_connect (channels, patterns, and shard_channels)

Added tests for binary channel and pattern reconnection in both sync
and async test suites.

Fixes #3912

* Applying review comments.

---------

Co-authored-by: petyaslavova <petya.slavova@redis.com>

Author: skylarkoo7

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

@@ -967,25 +967,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,54 @@ 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):
+            message = await wait_for_message(p)
+            assert message is not None
+            messages.append(message)
+
+        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):
+            message = await wait_for_message(p)
+            assert message is not None
+            messages.append(message)
+
+        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

@@ -201,6 +201,56 @@ 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):
+            message = wait_for_message(p)
+            assert message is not None
+            messages.append(message)
+
+        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):
+            message = wait_for_message(p)
+            assert message is not None
+            messages.append(message)
+
+        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
     ):