famedly · jason-famedly · Jul 10, 2026 · Jul 13, 2026 · Jul 13, 2026 · Aug 4, 2026
@@ -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`

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

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

@@ -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")
+        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(
             "Successfully subscribed to redis stream, sending REPLICATE command"
         )
         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

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