Skip to content
This repository has been archived by the owner on Apr 26, 2024. It is now read-only.

Commit

Permalink
Don't relay REMOTE_SERVER_UP cmds to same conn.
Browse files Browse the repository at this point in the history
For direct TCP connections we need the master to relay REMOTE_SERVER_UP
commands to the other connections so that all instances get notified
about it. The old implementation just relayed to all connections,
assuming that sending back to the original sender of the command was
safe. This is not true for redis, where commands sent get echoed back to
the sender, which was causing master to effectively infinite loop
sending and then re-receiving REMOTE_SERVER_UP commands that it sent.

The fix is to ensure that we only relay to *other* connections and not
to the connection we received the notification from.

Fixes #7334.
  • Loading branch information
erikjohnston committed Apr 27, 2020
1 parent 7022704 commit 659b6be
Show file tree
Hide file tree
Showing 2 changed files with 29 additions and 13 deletions.
9 changes: 0 additions & 9 deletions synapse/notifier.py
Original file line number Diff line number Diff line change
Expand Up @@ -220,12 +220,6 @@ def add_replication_callback(self, cb: Callable[[], None]):
"""
self.replication_callbacks.append(cb)

def add_remote_server_up_callback(self, cb: Callable[[str], None]):
"""Add a callback that will be called when synapse detects a server
has been
"""
self.remote_server_up_callbacks.append(cb)

def on_new_room_event(
self, event, room_stream_id, max_room_stream_id, extra_users=[]
):
Expand Down Expand Up @@ -544,6 +538,3 @@ def notify_remote_server_up(self, server: str):
# circular dependencies.
if self.federation_sender:
self.federation_sender.wake_destination(server)

for cb in self.remote_server_up_callbacks:
cb(server)
33 changes: 29 additions & 4 deletions synapse/replication/tcp/handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -115,7 +115,6 @@ def __init__(self, hs):
self._server_notices_sender = None
if self._is_master:
self._server_notices_sender = hs.get_server_notices_sender()
self._notifier.add_remote_server_up_callback(self.send_remote_server_up)

def start_replication(self, hs):
"""Helper method to start a replication connection to the remote server
Expand Down Expand Up @@ -344,8 +343,24 @@ async def on_REMOTE_SERVER_UP(
""""Called when get a new REMOTE_SERVER_UP command."""
self._replication_data_handler.on_remote_server_up(cmd.data)

if self._is_master:
self._notifier.notify_remote_server_up(cmd.data)
self._notifier.notify_remote_server_up(cmd.data)

# We relay to all other connections to ensure every instance gets the
# notification.
#
# When configured to use redis we'll always only have one connection and
# so this is a no-op (all instances will have already received the same
# REMOTE_SERVER_UP command).
#
# For direct TCP connections this will relay to all other connections
# connected to us. When on master this will correctly fan out to all
# other direct TCP clients and on workers there'll only be the one
# connection to master.
#
# (The logic here should also be sound if we have a mix of Redis and
# direct TCP connections so long as there is only one traffic route
# between two instances, but that is not currently supported).
self.send_command(cmd, ignore_conn=conn)

def new_connection(self, connection: AbstractConnection):
"""Called when we have a new connection.
Expand Down Expand Up @@ -390,11 +405,21 @@ def connected(self) -> bool:
"""
return bool(self._connections)

def send_command(self, cmd: Command):
def send_command(
self, cmd: Command, ignore_conn: Optional[AbstractConnection] = None
):
"""Send a command to all connected connections.
Args:
cmd
ignore_conn: If set don't send command to the given connection.
Used when relaying commands from one connection to all others.
"""
if self._connections:
for connection in self._connections:
if connection == ignore_conn:
continue

try:
connection.send_command(cmd)
except Exception:
Expand Down

0 comments on commit 659b6be

Please sign in to comment.