Add the `notify_appservices_from_worker` configuration option (superseding `notify_appservices`) to allow a generic worker to be designated as the worker to send traffic to Application Services. (#12452)

4 files changed, 109 insertions, 19 deletions

diff --git a/synapse/app/generic_worker.py b/synapse/app/generic_worker.py
index 1865c671f4..07dddc0b13 100644
--- a/synapse/app/generic_worker.py
+++ b/synapse/app/generic_worker.py
@@ -441,22 +441,6 @@ def start(config_options: List[str]) -> None:
         "synapse.app.user_dir",
     )
 
-    if config.worker.worker_app == "synapse.app.appservice":
-        if config.appservice.notify_appservices:
-            sys.stderr.write(
-                "\nThe appservices must be disabled in the main synapse process"
-                "\nbefore they can be run in a separate worker."
-                "\nPlease add ``notify_appservices: false`` to the main config"
-                "\n"
-            )
-            sys.exit(1)
-
-        # Force the appservice to start since they will be disabled in the main config
-        config.appservice.notify_appservices = True
-    else:
-        # For other worker types we force this to off.
-        config.appservice.notify_appservices = False
-
     if config.worker.worker_app == "synapse.app.user_dir":
         if config.server.update_user_directory:
             sys.stderr.write(
diff --git a/synapse/config/appservice.py b/synapse/config/appservice.py
index 720b90a283..b13cc6bb6e 100644
--- a/synapse/config/appservice.py
+++ b/synapse/config/appservice.py
@@ -33,7 +33,6 @@ class AppServiceConfig(Config):
 
     def read_config(self, config: JsonDict, **kwargs: Any) -> None:
         self.app_service_config_files = config.get("app_service_config_files", [])
-        self.notify_appservices = config.get("notify_appservices", True)
         self.track_appservice_user_ips = config.get("track_appservice_user_ips", False)
 
     def generate_config_section(cls, **kwargs: Any) -> str:
diff --git a/synapse/config/workers.py b/synapse/config/workers.py
index a5479dfca9..a9dbcc6d3d 100644
--- a/synapse/config/workers.py
+++ b/synapse/config/workers.py
@@ -14,7 +14,8 @@
 # limitations under the License.
 
 import argparse
-from typing import Any, List, Union
+import logging
+from typing import Any, Dict, List, Union
 
 import attr
 
@@ -42,6 +43,13 @@ synapse process before they can be run in a separate worker.
 Please add ``start_pushers: false`` to the main config
 """
 
+_DEPRECATED_WORKER_DUTY_OPTION_USED = """
+The '%s' configuration option is deprecated and will be removed in a future
+Synapse version. Please use ``%s: name_of_worker`` instead.
+"""
+
+logger = logging.getLogger(__name__)
+
 
 def _instance_to_list_converter(obj: Union[str, List[str]]) -> List[str]:
     """Helper for allowing parsing a string or list of strings to a config
@@ -296,6 +304,105 @@ class WorkerConfig(Config):
             self.worker_name is None and background_tasks_instance == "master"
         ) or self.worker_name == background_tasks_instance
 
+        self.should_notify_appservices = self._should_this_worker_perform_duty(
+            config,
+            legacy_master_option_name="notify_appservices",
+            legacy_worker_app_name="synapse.app.appservice",
+            new_option_name="notify_appservices_from_worker",
+        )
+
+    def _should_this_worker_perform_duty(
+        self,
+        config: Dict[str, Any],
+        legacy_master_option_name: str,
+        legacy_worker_app_name: str,
+        new_option_name: str,
+    ) -> bool:
+        """
+        Figures out whether this worker should perform a certain duty.
+
+        This function is temporary and is only to deal with the complexity
+        of allowing old, transitional and new configurations all at once.
+
+        Contradictions between the legacy and new part of a transitional configuration
+        will lead to a ConfigError.
+
+        Parameters:
+            config: The config dictionary
+            legacy_master_option_name: The name of a legacy option, whose value is boolean,
+                specifying whether it's the master that should handle a certain duty.
+                e.g. "notify_appservices"
+            legacy_worker_app_name: The name of a legacy Synapse worker application
+                that would traditionally perform this duty.
+                e.g. "synapse.app.appservice"
+            new_option_name: The name of the new option, whose value is the name of a
+                designated worker to perform the duty.
+                e.g. "notify_appservices_from_worker"
+        """
+
+        # None means 'unspecified'; True means 'run here' and False means
+        # 'don't run here'.
+        new_option_should_run_here = None
+        if new_option_name in config:
+            designated_worker = config[new_option_name] or "master"
+            new_option_should_run_here = (
+                designated_worker == "master" and self.worker_name is None
+            ) or designated_worker == self.worker_name
+
+        legacy_option_should_run_here = None
+        if legacy_master_option_name in config:
+            run_on_master = bool(config[legacy_master_option_name])
+
+            legacy_option_should_run_here = (
+                self.worker_name is None and run_on_master
+            ) or (self.worker_app == legacy_worker_app_name and not run_on_master)
+
+            # Suggest using the new option instead.
+            logger.warning(
+                _DEPRECATED_WORKER_DUTY_OPTION_USED,
+                legacy_master_option_name,
+                new_option_name,
+            )
+
+        if self.worker_app == legacy_worker_app_name and config.get(
+            legacy_master_option_name, True
+        ):
+            # As an extra bit of complication, we need to check that the
+            # specialised worker is only used if the legacy config says the
+            # master isn't performing the duties.
+            raise ConfigError(
+                f"Cannot use deprecated worker app type '{legacy_worker_app_name}' whilst deprecated option '{legacy_master_option_name}' is not set to false.\n"
+                f"Consider setting `worker_app: synapse.app.generic_worker` and using the '{new_option_name}' option instead.\n"
+                f"The '{new_option_name}' option replaces '{legacy_master_option_name}'."
+            )
+
+        if new_option_should_run_here is None and legacy_option_should_run_here is None:
+            # Neither option specified; the fallback behaviour is to run on the main process
+            return self.worker_name is None
+
+        if (
+            new_option_should_run_here is not None
+            and legacy_option_should_run_here is not None
+        ):
+            # Both options specified; ensure they match!
+            if new_option_should_run_here != legacy_option_should_run_here:
+                update_worker_type = (
+                    " and set worker_app: synapse.app.generic_worker"
+                    if self.worker_app == legacy_worker_app_name
+                    else ""
+                )
+                # If the values conflict, we suggest the admin removes the legacy option
+                # for simplicity.
+                raise ConfigError(
+                    f"Conflicting configuration options: {legacy_master_option_name} (legacy), {new_option_name} (new).\n"
+                    f"Suggestion: remove {legacy_master_option_name}{update_worker_type}.\n"
+                )
+
+        # We've already validated that these aren't conflicting; now just see if
+        # either is True.
+        # (By this point, these are either the same value or only one is not None.)
+        return bool(new_option_should_run_here or legacy_option_should_run_here)
+
     def generate_config_section(self, **kwargs: Any) -> str:
         return """\
         ## Workers ##
diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py
index b3894666cc..85bd5e4768 100644
--- a/synapse/handlers/appservice.py
+++ b/synapse/handlers/appservice.py
@@ -59,7 +59,7 @@ class ApplicationServicesHandler:
         self.scheduler = hs.get_application_service_scheduler()
         self.started_scheduler = False
         self.clock = hs.get_clock()
-        self.notify_appservices = hs.config.appservice.notify_appservices
+        self.notify_appservices = hs.config.worker.should_notify_appservices
         self.event_sources = hs.get_event_sources()
         self._msc2409_to_device_messages_enabled = (
             hs.config.experimental.msc2409_to_device_messages_enabled