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

Fix Redis reconnection logic #7482

Merged
erikjohnston merged 4 commits into from
May 13, 2020
Merged

Fix Redis reconnection logic #7482

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e27b607
Redis: only call new_connection after successful subscribe
erikjohnston May 12, 2020
ece0793
Send POSITIONs when connecting via Redis
erikjohnston May 12, 2020
6202254
Newsfile
erikjohnston May 12, 2020
cd13c66
Only send POSITION command to connection that sent REPLICATE
erikjohnston May 12, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions changelog.d/7482.bugfix
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Fix Redis reconnection logic that can result in missed updates over replication if master reconnects to Redis without restarting.
9 changes: 8 additions & 1 deletion synapse/replication/tcp/handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,14 +151,21 @@ def start_replication(self, hs):
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):

@clokep clokep May 12, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This doesn't seem to use conn at all. Is that on purpose?

@erikjohnston erikjohnston May 12, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, that's exciting! It should only be sending stuff to the given connection via conn.send_command but instead is sending it to all connections via self.send_command. It doesn't actually matter that we send POSITION spuriously to random connections, but lets fix it

"""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:
return

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)
)

Expand Down
7 changes: 6 additions & 1 deletion synapse/replication/tcp/redis.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,7 +70,6 @@ def connectionMade(self):
logger.info("Connected to redis")
super().connectionMade()
run_as_background_process("subscribe-replication", self._send_subscribe)
self.handler.new_connection(self)

async def _send_subscribe(self):
# it's important to make sure that we only send the REPLICATE command once we
Expand All @@ -81,9 +80,15 @@ async def _send_subscribe(self):
logger.info(
"Successfully subscribed to redis stream, sending REPLICATE command"
)
self.handler.new_connection(self)
await self._async_send_command(ReplicateCommand())
logger.info("REPLICATE successfully sent")

# We send out our positions when there is a new connection in case the
# other side missed updates. We do this for Redis connections as the
# otherside won't know we've connected and so won't issue a REPLICATE.
self.handler.send_positions_to_connection(self)

def messageReceived(self, pattern: str, channel: str, message: str):
"""Received a message from redis.
"""
Expand Down