diff --git a/docs/usage/configuration/config_documentation.md b/docs/usage/configuration/config_documentation.md index 90b001bcfe..47002379ea 100644 --- a/docs/usage/configuration/config_documentation.md +++ b/docs/usage/configuration/config_documentation.md @@ -111,6 +111,15 @@ Example configuration: famedly_maximum_refresh_token_lifetime: 3600000 ``` --- +### `force_crash_workers_after_main_restart` + +*(boolean)* When enabled, shutdown and exit all workers of Synapse if the main process is restarted. The shutdown is triggered by a SIGTERM raised from within each worker. Once shutdown, they are eligible for restart based on the deployment orchestrator. Individual workers can still be restarted independently without affecting other workers or the main process Defaults to `false`. + +Example configuration: +```yaml +force_crash_workers_after_main_restart: true +``` +--- ## Experimental features They all come from Element's Synapse except for `msc3706_enabled` diff --git a/schema/synapse-config.schema.yaml b/schema/synapse-config.schema.yaml index 78d664cf5f..2c7543dc9e 100644 --- a/schema/synapse-config.schema.yaml +++ b/schema/synapse-config.schema.yaml @@ -10,6 +10,16 @@ properties: default: "160d" examples: - 3600000 + force_crash_workers_after_main_restart: + type: boolean + description: >- + When enabled, shutdown and exit all workers of Synapse if the main process is restarted. The shutdown is triggered + by a SIGTERM raised from within each worker. Once shutdown, they are eligible for restart based on the deployment + orchestrator. + Individual workers can still be restarted independently without affecting other workers or the main process + default: false + examples: + - true experimental_features: type: object description: >- diff --git a/synapse/config/server.py b/synapse/config/server.py index 4dafeff0b7..2ac9eb876f 100644 --- a/synapse/config/server.py +++ b/synapse/config/server.py @@ -983,6 +983,13 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None: self.max_delayed_events_per_user and self.max_event_delay_duration ) + 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..cd80f7c150 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,14 +151,32 @@ 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") + await self._async_send_command( + ServerCommand( + self.server_name, + self.hs.get_instance_name(), + self.synapse_handler.startup_time_ms, + ) + ) + # 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( @@ -165,6 +184,7 @@ async def _send_subscribe(self) -> None: ) self.synapse_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 diff --git a/tests/replication/tcp/test_handler.py b/tests/replication/tcp/test_handler.py index d35191e654..d34100b310 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 @@ -221,3 +226,62 @@ 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 + """ + # This test simulates the process being restarted as close as we can from inside + # pytest/trial. The test infrastructure does not tolerate an actual SIGTERM + # being raised, as that will stop all tests in progress. Instead, we mock the + # `raise_signal` function to make sure it is actually called after the SERVER + # command has been sent. + + # We will need another worker that can receive the command + worker = self.make_worker_hs( + "synapse.app.generic_worker", + extra_config={ + "worker_name": "worker1", + "redis": {"enabled": True}, + }, + ) + + # + # Move forward time, so the SERVER command has a new value. A healthy amount, + # although a tiny amount would also do fine. + self.reactor.advance(1.000) + + # 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) + + # the `raise_signal` should have been called exactly one time. + patch_ctx.assert_called_once()