diff --git a/synapse/config/server.py b/synapse/config/server.py index 3b57531fa4..258765b422 100644 --- a/synapse/config/server.py +++ b/synapse/config/server.py @@ -918,6 +918,13 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None: else: self.max_event_delay_ms = None + force_crash_workers_after_main_restart = config.get( + "force_crash_workers_after_main_restart", False + ) + self.force_crash_workers_after_main_restart = bool( + force_crash_workers_after_main_restart + ) + def has_tls_listener(self) -> bool: return any(listener.is_tls() for listener in self.listeners) diff --git a/synapse/replication/tcp/commands.py b/synapse/replication/tcp/commands.py index 0c85fb36fc..2efde4010b 100644 --- a/synapse/replication/tcp/commands.py +++ b/synapse/replication/tcp/commands.py @@ -95,16 +95,31 @@ def to_line(self) -> str: return self.data -class ServerCommand(_SimpleCommand): +class ServerCommand(Command): """Sent by the server on new connection and includes the server_name. Format:: - SERVER + SERVER """ NAME = "SERVER" + def __init__(self, server_name: str, instance_name: str, startup_time_ms: int): + self.server_name = server_name + self.instance_name = instance_name + self.startup_time_ms = startup_time_ms + + @classmethod + def from_line(cls: type["ServerCommand"], line: str) -> "ServerCommand": + server_name, instance_name, startup_time_ms = line.split(" ", 2) + return cls(server_name, instance_name, int(startup_time_ms)) + + def to_line(self) -> str: + return " ".join( + (self.server_name, self.instance_name, str(self.startup_time_ms)) + ) + class RdataCommand(Command): """Sent by server when a subscribed stream has an update. diff --git a/synapse/replication/tcp/handler.py b/synapse/replication/tcp/handler.py index ad9fed72dd..6f4385de4f 100644 --- a/synapse/replication/tcp/handler.py +++ b/synapse/replication/tcp/handler.py @@ -20,6 +20,7 @@ # # import logging +import signal from typing import ( TYPE_CHECKING, Any, @@ -33,6 +34,7 @@ from twisted.internet.protocol import ReconnectingClientFactory +from synapse.config.workers import MAIN_PROCESS_INSTANCE_NAME from synapse.metrics import SERVER_NAME_LABEL, LaterGauge from synapse.replication.tcp.commands import ( CancelTaskCommand, @@ -45,6 +47,7 @@ RdataCommand, RemoteServerUpCommand, ReplicateCommand, + ServerCommand, UserIpCommand, UserSyncCommand, ) @@ -133,6 +136,8 @@ def __init__(self, hs: "HomeServer"): self._clock = hs.get_clock() self._instance_id = hs.get_instance_id() self._instance_name = hs.get_instance_name() + self.force_restart = hs.config.server.force_crash_workers_after_main_restart + self.startup_time_ms = self._clock.time_msec() # Additional Redis channel suffixes to subscribe to. self._channels_to_subscribe_to: list[str] = [] @@ -442,6 +447,32 @@ def get_streams_to_replicate(self) -> list[Stream]: """Get a list of streams that this instances replicates.""" return self._streams_to_replicate + def on_SERVER(self, conn: IReplicationConnection, cmd: ServerCommand) -> None: + if cmd.server_name != self.server_name: + logger.error( + "Current process is receiving commands from wrong remote: %r", + cmd.server_name, + ) + + # Simply: + # * if force restarting is enabled + # * and if this worker process is not the main process + # * and the command came from the main process + # * and the timestamp from the command is >= to the timestamp this worker + # started at + # ... crash the worker + if ( + self.force_restart + and self._instance_name != MAIN_PROCESS_INSTANCE_NAME + and cmd.instance_name == MAIN_PROCESS_INSTANCE_NAME + and cmd.startup_time_ms >= self.startup_time_ms + ): + logger.error( + "Main process broadcast it just came up, crashing local worker" + ) + + signal.raise_signal(signal.SIGTERM) + def on_REPLICATE(self, conn: IReplicationConnection, cmd: ReplicateCommand) -> None: self.send_positions_to_connection() diff --git a/synapse/replication/tcp/protocol.py b/synapse/replication/tcp/protocol.py index 489a2c76a6..6ca1774ce2 100644 --- a/synapse/replication/tcp/protocol.py +++ b/synapse/replication/tcp/protocol.py @@ -26,6 +26,7 @@ import fcntl import logging +import signal import struct from inspect import isawaitable from typing import TYPE_CHECKING, Any, Collection @@ -38,6 +39,7 @@ from twisted.protocols.basic import LineOnlyReceiver from twisted.python.failure import Failure +from synapse.config.workers import MAIN_PROCESS_INSTANCE_NAME from synapse.logging.context import PreserveLoggingContext from synapse.metrics import SERVER_NAME_LABEL, LaterGauge from synapse.metrics.background_process_metrics import ( @@ -485,15 +487,40 @@ def __init__( super().__init__(hs, server_name, clock, handler) self.server_name = server_name + self.instance_name = hs.get_instance_name() + self.clock = hs.get_clock() + self.force_restart = hs.config.server.force_crash_workers_after_main_restart + self.startup_time_ms = self.clock.time_msec() def connectionMade(self) -> None: - self.send_command(ServerCommand(self.server_name)) + self.send_command( + ServerCommand(self.server_name, self.instance_name, self.startup_time_ms) + ) super().connectionMade() def on_NAME(self, cmd: NameCommand) -> None: logger.info("[%s] Renamed to %r", self.id(), cmd.data) self.name = cmd.data + def on_SERVER(self, cmd: ServerCommand) -> None: + if cmd.server_name != self.server_name: + logger.error( + "[%s] Connected to wrong remote: %r", self.id(), cmd.server_name + ) + self.send_error("Wrong remote") + if ( + self.force_restart + and self.instance_name != MAIN_PROCESS_INSTANCE_NAME + and cmd.instance_name == MAIN_PROCESS_INSTANCE_NAME + and cmd.startup_time_ms >= self.startup_time_ms + ): + logger.error( + "Main process broadcast it just came up, crashing local worker" + ) + self.send_error("Terminating local worker") + + signal.raise_signal(signal.SIGTERM) + class ClientReplicationStreamProtocol(BaseReplicationStreamProtocol): VALID_INBOUND_COMMANDS = VALID_SERVER_COMMANDS @@ -520,8 +547,10 @@ def connectionMade(self) -> None: self.replicate() def on_SERVER(self, cmd: ServerCommand) -> None: - if cmd.data != self.server_name: - logger.error("[%s] Connected to wrong remote: %r", self.id(), cmd.data) + if cmd.server_name != self.server_name: + logger.error( + "[%s] Connected to wrong remote: %r", self.id(), cmd.server_name + ) self.send_error("Wrong remote") def replicate(self) -> None: diff --git a/synapse/replication/tcp/redis.py b/synapse/replication/tcp/redis.py index 93ba48b406..fe05a3d59e 100644 --- a/synapse/replication/tcp/redis.py +++ b/synapse/replication/tcp/redis.py @@ -45,6 +45,7 @@ from synapse.replication.tcp.commands import ( Command, ReplicateCommand, + ServerCommand, parse_command_from_line, ) from synapse.replication.tcp.context import ClientContextFactory @@ -150,21 +151,40 @@ def connectionMade(self) -> None: self.hs.run_as_background_process("subscribe-replication", self._send_subscribe) async def _send_subscribe(self) -> None: + # Make sure to send the SERVER command before any other commands. It is + # important to not interrupt the SUBSCRIBE processes + logger.info("Sending SERVER command to redis connection") + self.send_command( + ServerCommand( + self.server_name, + self.hs.get_instance_name(), + self.hs.get_clock().time_msec(), + ) + ) + # 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. + # XXX: Oddly, other places in the code specify that commands are not to be sent + # on a connection that has already been subscribed to and another connection is + # set up for this purpose. Isn't sending a replicate command on a subscribed + # stream exactly the opposite of this? fully_qualified_stream_names = [ f"{self.synapse_stream_prefix}/{stream_suffix}" for stream_suffix in self.synapse_channel_names ] + [self.synapse_stream_prefix] - logger.info("Sending redis SUBSCRIBE for %r", fully_qualified_stream_names) + logger.info( + "Successfully sent SERVER command, sending 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" ) self.synapse_handler.new_connection(self) - await self._async_send_command(ReplicateCommand()) + self.send_command(ReplicateCommand()) + logger.info("REPLICATE successfully sent") # We send out our positions when there is a new connection in case the diff --git a/tests/replication/tcp/test_handler.py b/tests/replication/tcp/test_handler.py index a8eb7fc523..d16be51939 100644 --- a/tests/replication/tcp/test_handler.py +++ b/tests/replication/tcp/test_handler.py @@ -18,10 +18,15 @@ # [This file includes modifications made by New Vector Limited] # # +from typing import Any +from unittest import mock from twisted.internet import defer +from twisted.internet.testing import MemoryReactor -from synapse.replication.tcp.commands import PositionCommand +from synapse.replication.tcp.commands import PositionCommand, ServerCommand +from synapse.server import HomeServer +from synapse.util.clock import Clock from tests.replication._base import BaseMultiWorkerStreamTestCase @@ -209,3 +214,52 @@ def test_wait_for_stream_position_rdata(self) -> None: # Master should get told about `next_token2`, so the deferred should # resolve. self.assertTrue(d.called) + + +class RestartBehaviorTestCase(BaseMultiWorkerStreamTestCase): + def prepare( + self, reactor: MemoryReactor, clock: Clock, homeserver: HomeServer + ) -> None: + super().prepare(reactor, clock, homeserver) + self.repl_comm_handler = self.hs.get_replication_command_handler() + + self.server_name = self.hs.config.server.server_name + + def default_config(self) -> dict[str, Any]: + config = super().default_config() + config["force_crash_workers_after_main_restart"] = True + return config + + def test_server_command_can_raise_from_main(self) -> None: + """ + Test that a server command send from the main process can raise a SIGTERM on a + worker + """ + # self.reactor.advance(0.001) + worker = self.make_worker_hs( + "synapse.app.generic_worker", + extra_config={ + "worker_name": "worker1", + "redis": {"enabled": True}, + }, + ) + + # Move forward time just a pinch, so the SERVER command has a new value + self.reactor.advance(0.001) + # In unit tests, we can not really restart a process. So, send the command again + # to simulate that. + self.repl_comm_handler.send_command( + ServerCommand( + self.server_name, self.hs.get_instance_name(), self.clock.time_msec() + ) + ) + with mock.patch( + "signal.raise_signal", + ) as patch_ctx: + # Simulate receiving the redis command by calling the handler that would + # have responded to "wake up" + worker.get_replication_streamer().on_notifier_poke() + # Need to move forward time to get the background tasks to run + self.reactor.advance(0.001) + + patch_ctx.assert_called_once()