Lay some foundation work to allow workers to only subscribe to some kinds of messages, reducing replication traffic.

changelog.d/12672.misc

@@ -0,0 +1 @@
+Lay some foundation work to allow workers to only subscribe to some kinds of messages, reducing replication traffic.

synapse/replication/tcp/redis.py

@@ -14,7 +14,7 @@
 
 import logging
 from inspect import isawaitable
-from typing import TYPE_CHECKING, Any, Generic, Optional, Type, TypeVar, cast
+from typing import TYPE_CHECKING, Any, Generic, List, Optional, Type, TypeVar, cast
 
 import attr
 import txredisapi
@@ -24,6 +24,7 @@
 from twisted.internet.interfaces import IAddress, IConnector
 from twisted.python.failure import Failure
 
+from synapse.config.homeserver import HomeServerConfig
 from synapse.logging.context import PreserveLoggingContext, make_deferred_yieldable
 from synapse.metrics.background_process_metrics import (
     BackgroundProcessLoggingContext,
@@ -85,14 +86,15 @@ class RedisSubscriber(txredisapi.SubscriberProtocol):
 
     Attributes:
         synapse_handler: The command handler to handle incoming commands.
-        synapse_stream_name: The *redis* stream name to subscribe to and publish
+        synapse_stream_prefix: The *redis* stream name to subscribe to and publish
             from (not anything to do with Synapse replication streams).
         synapse_outbound_redis_connection: The connection to redis to use to send
             commands.
     """
 
     synapse_handler: "ReplicationCommandHandler"
-    synapse_stream_name: str
+    synapse_stream_prefix: str
+    synapse_subscribed_channels: List[str]
     synapse_outbound_redis_connection: txredisapi.ConnectionHandler
 
     def __init__(self, *args: Any, **kwargs: Any):
@@ -117,8 +119,13 @@ async def _send_subscribe(self) -> None:
         # it's important to make sure that we only send the REPLICATE command once we
         # have successfully subscribed to the stream - otherwise we might miss the
         # POSITION response sent back by the other end.
-        logger.info("Sending redis SUBSCRIBE for %s", self.synapse_stream_name)
-        await make_deferred_yieldable(self.subscribe(self.synapse_stream_name))
+        fully_qualified_stream_names = [
+            f"{self.synapse_stream_prefix}/{stream_suffix}"
+            for stream_suffix in self.synapse_subscribed_channels
+        ] + [self.synapse_stream_prefix]
+        logger.info("Sending redis SUBSCRIBE for %r", fully_qualified_stream_names)
+        await make_deferred_yieldable(self.subscribe(fully_qualified_stream_names))
+
         logger.info(
             "Successfully subscribed to redis stream, sending REPLICATE command"
         )
@@ -217,7 +224,7 @@ async def _async_send_command(self, cmd: Command) -> None:
 
         await make_deferred_yieldable(
             self.synapse_outbound_redis_connection.publish(
-                self.synapse_stream_name, encoded_string
+                self.synapse_stream_prefix, encoded_string
             )
         )
 
@@ -300,7 +307,7 @@ def format_address(address: IAddress) -> str:
 
 class RedisDirectTcpReplicationClientFactory(SynapseRedisFactory):
     """This is a reconnecting factory that connects to redis and immediately
-    subscribes to a stream.
+    subscribes to some streams.
 
     Args:
         hs
@@ -326,10 +333,47 @@ def __init__(
         )
 
         self.synapse_handler = hs.get_replication_command_handler()
-        self.synapse_stream_name = hs.hostname
+        self.synapse_stream_prefix = hs.hostname
+        self.synapse_subscribed_channels = (
+            RedisDirectTcpReplicationClientFactory.channels_to_subscribe_to_for_config(
+                hs.config
+            )
+        )
 
         self.synapse_outbound_redis_connection = outbound_redis_connection
 
+    @staticmethod
+    def channels_to_subscribe_to_for_config(config: HomeServerConfig) -> List[str]:
+        subscribe_to = []
+
+        if config.worker.run_background_tasks or config.worker.worker_app is None:
+            # If we're the main process or the background worker, we want to process
+            # User IP addresses
+            subscribe_to.append("USER_IP")
+
+        # Subscribe to the following RDATA channels.
+        # We may be able to reduce this in the future.
+        subscribe_to += [
+            "RDATA/account_data",
+            "RDATA/backfill",
+            "RDATA/caches",
+            "RDATA/device_lists",
+            "RDATA/events",
+            "RDATA/federation",
+            "RDATA/groups",
+            "RDATA/presence",
+            "RDATA/presence_federation",
+            "RDATA/push_rules",
+            "RDATA/pushers",
+            "RDATA/receipts",
+            "RDATA/tag_account_data",
+            "RDATA/to_device",
+            "RDATA/typing",
+            "RDATA/user_signature",
+        ]
+
+        return subscribe_to
+
     def buildProtocol(self, addr: IAddress) -> RedisSubscriber:
         p = super().buildProtocol(addr)
         p = cast(RedisSubscriber, p)
@@ -340,7 +384,8 @@ def buildProtocol(self, addr: IAddress) -> RedisSubscriber:
         # protocol.
         p.synapse_handler = self.synapse_handler
         p.synapse_outbound_redis_connection = self.synapse_outbound_redis_connection
-        p.synapse_stream_name = self.synapse_stream_name
+        p.synapse_stream_prefix = self.synapse_stream_prefix
+        p.synapse_subscribed_channels = self.synapse_subscribed_channels
 
         return p
 

tests/replication/_base.py

@@ -12,7 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import logging
-from typing import Any, Dict, List, Optional, Tuple
+from collections import defaultdict
+from typing import Any, Dict, List, Optional, Set, Tuple
 
 from twisted.internet.address import IPv4Address
 from twisted.internet.protocol import Protocol
@@ -475,22 +476,25 @@ class FakeRedisPubSubServer:
     """A fake Redis server for pub/sub."""
 
     def __init__(self):
-        self._subscribers = set()
+        self._subscribers_by_channel: Dict[
+            bytes, Set["FakeRedisPubSubProtocol"]
+        ] = defaultdict(set)
 
-    def add_subscriber(self, conn):
+    def add_subscriber(self, conn, channel: bytes):
         """A connection has called SUBSCRIBE"""
-        self._subscribers.add(conn)
+        self._subscribers_by_channel[channel].add(conn)
 
     def remove_subscriber(self, conn):
-        """A connection has called UNSUBSCRIBE"""
-        self._subscribers.discard(conn)
+        """A connection has lost connection"""
+        for subscribers in self._subscribers_by_channel.values():
+            subscribers.discard(conn)
 
-    def publish(self, conn, channel, msg) -> int:
+    def publish(self, conn, channel: bytes, msg) -> int:
         """A connection want to publish a message to subscribers."""
-        for sub in self._subscribers:
+        for sub in self._subscribers_by_channel[channel]:
             sub.send(["message", channel, msg])
 
-        return len(self._subscribers)
+        return len(self._subscribers_by_channel)
 
     def buildProtocol(self, addr):
         return FakeRedisPubSubProtocol(self)
@@ -531,9 +535,10 @@ def handle_command(self, command, *args):
             num_subscribers = self._server.publish(self, channel, message)
             self.send(num_subscribers)
         elif command == b"SUBSCRIBE":
-            (channel,) = args
-            self._server.add_subscriber(self)
-            self.send(["subscribe", channel, 1])
+            for idx, channel in enumerate(args):
+                num_channels = idx + 1
+                self._server.add_subscriber(self, channel)
+                self.send(["subscribe", channel, num_channels])
 
         # Since we use SET/GET to cache things we can safely no-op them.
         elif command == b"SET":

tests/replication/tcp/test_redis.py

@@ -0,0 +1,103 @@
+# Copyright 2022 The Matrix.org Foundation C.I.C.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+try:
+    # We only import it to see if it's installed, so ignore the 'unused' import
+    import txredisapi  # noqa: F401
+
+    HAVE_TXREDISAPI = True
+except ImportError:
+    HAVE_TXREDISAPI = False
+
+from tests.replication._base import BaseMultiWorkerStreamTestCase
+from tests.unittest import HomeserverTestCase
+
+ALL_RDATA_CHANNELS = [
+    "RDATA/account_data",
+    "RDATA/backfill",
+    "RDATA/caches",
+    "RDATA/device_lists",
+    "RDATA/events",
+    "RDATA/federation",
+    "RDATA/groups",
+    "RDATA/presence",
+    "RDATA/presence_federation",
+    "RDATA/push_rules",
+    "RDATA/pushers",
+    "RDATA/receipts",
+    "RDATA/tag_account_data",
+    "RDATA/to_device",
+    "RDATA/typing",
+    "RDATA/user_signature",
+]
+
+
+class RedisTestCase(HomeserverTestCase):
+    if not HAVE_TXREDISAPI:
+        skip = "Redis extras not installed"
+
+    def test_subscribed_to_enough_redis_channels(self) -> None:
+        from synapse.replication.tcp.redis import RedisDirectTcpReplicationClientFactory
+
+        # The default main process is subscribed to USER_IP and all RDATA channels.
+        self.assertCountEqual(
+            RedisDirectTcpReplicationClientFactory.channels_to_subscribe_to_for_config(
+                self.hs.config
+            ),
+            [
+                "USER_IP",
+            ]
+            + ALL_RDATA_CHANNELS,
+        )
+
+
+class RedisWorkerTestCase(BaseMultiWorkerStreamTestCase):
+    if not HAVE_TXREDISAPI:
+        skip = "Redis extras not installed"
+
+    def test_background_worker_subscribed_to_user_ip(self) -> None:
+        from synapse.replication.tcp.redis import RedisDirectTcpReplicationClientFactory
+
+        # The default main process is subscribed to USER_IP and all RDATA channels.
+        worker1 = self.make_worker_hs(
+            "synapse.app.generic_worker",
+            extra_config={
+                "worker_name": "worker1",
+                "run_background_tasks_on": "worker1",
+            },
+        )
+        self.assertIn(
+            "USER_IP",
+            RedisDirectTcpReplicationClientFactory.channels_to_subscribe_to_for_config(
+                worker1.config
+            ),
+        )
+
+    def test_non_background_worker_not_subscribed_to_user_ip(self) -> None:
+        from synapse.replication.tcp.redis import RedisDirectTcpReplicationClientFactory
+
+        # The default main process is subscribed to USER_IP and all RDATA channels.
+        worker2 = self.make_worker_hs(
+            "synapse.app.generic_worker",
+            extra_config={
+                "worker_name": "worker2",
+                "run_background_tasks_on": "worker1",
+            },
+        )
+        self.assertNotIn(
+            "USER_IP",
+            RedisDirectTcpReplicationClientFactory.channels_to_subscribe_to_for_config(
+                worker2.config
+            ),
+        )