Restructure the way we subscribe to channels

Author: reivilibre

synapse/replication/tcp/handler.py

@@ -1,5 +1,5 @@
 # Copyright 2017 Vector Creations Ltd
-# Copyright 2020 The Matrix.org Foundation C.I.C.
+# Copyright 2020, 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.
@@ -101,6 +101,9 @@ def __init__(self, hs: "HomeServer"):
         self._instance_id = hs.get_instance_id()
         self._instance_name = hs.get_instance_name()
 
+        # Additional Redis channel suffixes to subscribe to.
+        self._channels_to_subscribe_to: List[str] = []
+
         self._is_presence_writer = (
             hs.get_instance_name() in hs.config.worker.writers.presence
         )
@@ -243,6 +246,31 @@ def __init__(self, hs: "HomeServer"):
             # If we're NOT using Redis, this must be handled by the master
             self._should_insert_client_ips = hs.get_instance_name() == "master"
 
+        if self._is_master or self._should_insert_client_ips:
+            self.subscribe_to_channel("USER_IP")
+
+    def subscribe_to_channel(self, channel_name: str) -> None:
+        """
+        Indicates that we wish to subscribe to a Redis channel by name.
+
+        (The name will later be prefixed with the server name; i.e. subscribing
+        to the 'ABC' channel actually subscribes to 'example.com/ABC' Redis-side.)
+
+        Raises:
+          - If replication has already started, then it's too late to subscribe
+            to new channels.
+        """
+
+        if self._factory is not None:
+            # We don't allow subscribing after the fact to avoid the chance
+            # of missing an important message because we didn't subscribe in time.
+            raise RuntimeError(
+                "Cannot subscribe to more channels after replication started."
+            )
+
+        if channel_name not in self._channels_to_subscribe_to:
+            self._channels_to_subscribe_to.append(channel_name)
+
     def _add_command_to_stream_queue(
         self, conn: IReplicationConnection, cmd: Union[RdataCommand, PositionCommand]
     ) -> None:
@@ -323,9 +351,7 @@ def start_replication(self, hs: "HomeServer") -> None:
             self._factory = RedisDirectTcpReplicationClientFactory(
                 hs,
                 outbound_redis_connection,
-                channel_names=RedisDirectTcpReplicationClientFactory.channels_to_subscribe_to_for_config(
-                    hs.config
-                ),
+                channel_names=self._channels_to_subscribe_to,
             )
             hs.get_reactor().connectTCP(
                 hs.config.redis.redis_host,

synapse/replication/tcp/redis.py

@@ -24,7 +24,6 @@
 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,
@@ -345,17 +344,6 @@ def __init__(
 
         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")
-
-        return subscribe_to
-
     def buildProtocol(self, addr: IAddress) -> RedisSubscriber:
         p = super().buildProtocol(addr)
         p = cast(RedisSubscriber, p)