famedly · jason-famedly · Jul 10, 2026 · Jul 13, 2026 · Jul 13, 2026 · FrenchGithubUser
@@ -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)
 

@@ -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 <server_name> <instance_name> <startup_time_ms>
     """
 
     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.

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

@@ -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:

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

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