summary refs log tree commit diff
path: root/synapse/replication/tcp/handler.py
diff options
context:
space:
mode:
authorErik Johnston <erik@matrix.org>2020-05-13 09:57:15 +0100
committerGitHub <noreply@github.com>2020-05-13 09:57:15 +0100
commit8ca79613e62201f0990a482130527fc88c4ffd40 (patch)
tree165ff43a8cf7bd954ac6b0ec6e65bf5db5fe5edc /synapse/replication/tcp/handler.py
parentUpdate documentation about SSO mapping providers (#7458) (diff)
downloadsynapse-8ca79613e62201f0990a482130527fc88c4ffd40.tar.xz
Fix Redis reconnection logic (#7482)
Proactively send out `POSITION` commands (as if we had just received a `REPLICATE`) when we connect to Redis. This is important as other instances won't notice we've connected to issue a `REPLICATE` command (unlike for direct TCP connections). This is only currently an issue if master process reconnects without restarting (if it restarts then it won't have written anything and so other instances probably won't have missed anything). 
Diffstat (limited to 'synapse/replication/tcp/handler.py')
-rw-r--r--synapse/replication/tcp/handler.py9
1 files changed, 8 insertions, 1 deletions
diff --git a/synapse/replication/tcp/handler.py b/synapse/replication/tcp/handler.py
index 1b05468483..e6a50aa74e 100644
--- a/synapse/replication/tcp/handler.py
+++ b/synapse/replication/tcp/handler.py
@@ -151,6 +151,13 @@ class ReplicationCommandHandler:
             hs.get_reactor().connectTCP(host, port, self._factory)
 
     async def on_REPLICATE(self, conn: AbstractConnection, cmd: ReplicateCommand):
+        self.send_positions_to_connection(conn)
+
+    def send_positions_to_connection(self, conn: AbstractConnection):
+        """Send current position of all streams this process is source of to
+        the connection.
+        """
+
         # We only want to announce positions by the writer of the streams.
         # Currently this is just the master process.
         if not self._is_master:
@@ -158,7 +165,7 @@ class ReplicationCommandHandler:
 
         for stream_name, stream in self._streams.items():
             current_token = stream.current_token(self._instance_name)
-            self.send_command(
+            conn.send_command(
                 PositionCommand(stream_name, self._instance_name, current_token)
             )