Replace the synapse.app.appservice and synapse.app.user_dir worker types with more flexible config options

changelog.d/10616.feature

@@ -0,0 +1 @@
+Allow a single worker to both notify appservices and update the user directory by removing the dependence on the `worker_type` worker config option.

docs/sample_config.yaml

@@ -2761,11 +2761,23 @@ opentracing:
 #  events: worker1
 #  typing: worker1
 
-# The worker that is used to run background tasks (e.g. cleaning up expired
-# data). If not provided this defaults to the main process.
+# The name of the worker that is used to run background tasks (e.g. cleaning
+# up expired data). If not provided this defaults to the main process.
 #
 #run_background_tasks_on: worker1
 
+# The name of the worker that is used to notify application services of new
+# traffic within their configured namespace. If not provided this defaults
+# to the main process.
+#
+#notify_appservices_from_worker: worker2
+
+# The name of the worker that is used to update the user directory tables as
+# users are created, update their room memberships as well as their profiles.
+# If not provided this defaults to the main process.
+#
+#update_user_directory_on_worker: worker3
+
 # A shared secret used by the replication APIs to authenticate HTTP requests
 # from workers.
 #

synapse/app/admin_cmd.py

@@ -197,7 +197,7 @@ def start(config_options):
         config.no_redirect_stdio = True
 
     # Explicitly disable background processes
-    config.update_user_directory = False
+    config.worker.should_update_user_directory = False
     config.run_background_tasks = False
     config.start_pushers = False
     config.pusher_shard_config.instances = []

synapse/app/generic_worker.py

@@ -427,38 +427,6 @@ def start(config_options):
         "synapse.app.user_dir",
     )
 
-    if config.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_app == "synapse.app.user_dir":
-        if config.server.update_user_directory:
-            sys.stderr.write(
-                "\nThe update_user_directory must be disabled in the main synapse process"
-                "\nbefore they can be run in a separate worker."
-                "\nPlease add ``update_user_directory: false`` to the main config"
-                "\n"
-            )
-            sys.exit(1)
-
-        # Force the pushers to start since they will be disabled in the main config
-        config.server.update_user_directory = True
-    else:
-        # For other worker types we force this to off.
-        config.server.update_user_directory = False
-
     synapse.events.USE_FROZEN_DICTS = config.use_frozen_dicts
     synapse.util.caches.TRACK_MEMORY_USAGE = config.caches.track_memory_usage
 

synapse/config/appservice.py

@@ -32,7 +32,6 @@ class AppServiceConfig(Config):
 
     def read_config(self, config, **kwargs):
         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):

synapse/config/server.py

@@ -257,10 +257,6 @@ def read_config(self, config, **kwargs):
                 self.presence_router_config,
             ) = load_module(presence_router_config, ("presence", "presence_router"))
 
-        # Whether to update the user directory or not. This should be set to
-        # false only if we are updating the user directory in a worker
-        self.update_user_directory = config.get("update_user_directory", True)
-
         # whether to enable the media repository endpoints. This should be set
         # to false if the media repository is running as a separate endpoint;
         # doing so ensures that we will not run cache cleanup jobs on the

synapse/config/workers.py

@@ -275,13 +275,39 @@ def read_config(self, config, **kwargs):
         # be able to run on only a single instance (meaning that they don't
         # depend on any in-memory state of a particular worker).
         #
-        # No effort is made to ensure only a single instance of these tasks is
-        # running.
+        # No effort is made here to ensure only a single instance of these tasks
+        # are running.
         background_tasks_instance = config.get("run_background_tasks_on") or "master"
         self.run_background_tasks = (
             self.worker_name is None and background_tasks_instance == "master"
         ) or self.worker_name == background_tasks_instance
 
+        # Whether this worker should notify appservices of traffic within their namespace.
+        #
+        # As a note for developers, this task is currently not shardable, and thus should
+        # only be handled by a single process.
+        #
+        # No effort is made here to ensure only a single instance of this task running.
+        notify_appservices_instance = (
+            config.get("notify_appservices_from_worker") or "master"
+        )
+        self.should_notify_appservices = (
+            self.worker_name is None and notify_appservices_instance == "master"
+        ) or self.worker_name == notify_appservices_instance
+
+        # Whether this worker should update the user directory tables.
+        #
+        # As a note for developers, this task is currently not shardable, and thus should
+        # only be handled by a single process.
+        #
+        # No effort is made here to ensure only a single instance of this task running.
+        update_user_directory_instance = (
+            config.get("update_user_directory_on_worker") or "master"
+        )
+        self.should_update_user_directory = (
+            self.worker_name is None and update_user_directory_instance == "master"
+        ) or self.worker_name == update_user_directory_instance
+
     def generate_config_section(self, config_dir_path, server_name, **kwargs):
         return """\
         ## Workers ##
@@ -318,11 +344,23 @@ def generate_config_section(self, config_dir_path, server_name, **kwargs):
         #  events: worker1
         #  typing: worker1
 
-        # The worker that is used to run background tasks (e.g. cleaning up expired
-        # data). If not provided this defaults to the main process.
+        # The name of the worker that is used to run background tasks (e.g. cleaning
+        # up expired data). If not provided this defaults to the main process.
         #
         #run_background_tasks_on: worker1
 
+        # The name of the worker that is used to notify application services of new
+        # traffic within their configured namespace. If not provided this defaults
+        # to the main process.
+        #
+        #notify_appservices_from_worker: worker2
+
+        # The name of the worker that is used to update the user directory tables as
+        # users are created, update their room memberships as well as their profiles.
+        # If not provided this defaults to the main process.
+        #
+        #update_user_directory_on_worker: worker3
+
         # A shared secret used by the replication APIs to authenticate HTTP requests
         # from workers.
         #

synapse/handlers/appservice.py

@@ -52,7 +52,7 @@ def __init__(self, hs: "HomeServer"):
         self.scheduler = hs.get_application_service_scheduler()
         self.started_scheduler = False
         self.clock = hs.get_clock()
-        self.notify_appservices = hs.config.notify_appservices
+        self.notify_appservices = hs.config.worker.should_notify_appservices
         self.event_sources = hs.get_event_sources()
 
         self.current_max = 0

synapse/handlers/user_directory.py

@@ -48,7 +48,7 @@ def __init__(self, hs: "HomeServer"):
         self.clock = hs.get_clock()
         self.notifier = hs.get_notifier()
         self.is_mine_id = hs.is_mine_id
-        self.update_user_directory = hs.config.update_user_directory
+        self.update_user_directory = hs.config.worker.should_update_user_directory
         self.search_all_users = hs.config.user_directory_search_all_users
         self.spam_checker = hs.get_spam_checker()
         # The current position in the current_state_delta stream

synapse/rest/client/v2_alpha/shared_rooms.py

@@ -36,7 +36,7 @@ def __init__(self, hs):
         super().__init__()
         self.auth = hs.get_auth()
         self.store = hs.get_datastore()
-        self.user_directory_active = hs.config.update_user_directory
+        self.user_directory_active = hs.config.worker.should_update_user_directory
 
     async def on_GET(self, request, user_id):
 

tests/handlers/test_user_directory.py

@@ -40,7 +40,6 @@ class UserDirectoryTestCase(unittest.HomeserverTestCase):
     def make_homeserver(self, reactor, clock):
 
         config = self.default_config()
-        config["update_user_directory"] = True
         return self.setup_test_homeserver(config=config)
 
     def prepare(self, reactor, clock, hs):
@@ -653,7 +652,6 @@ class TestUserDirSearchDisabled(unittest.HomeserverTestCase):
 
     def make_homeserver(self, reactor, clock):
         config = self.default_config()
-        config["update_user_directory"] = True
         hs = self.setup_test_homeserver(config=config)
 
         self.config = hs.config

tests/rest/client/v2_alpha/test_shared_rooms.py

@@ -33,7 +33,6 @@ class UserSharedRoomsTest(unittest.HomeserverTestCase):
 
     def make_homeserver(self, reactor, clock):
         config = self.default_config()
-        config["update_user_directory"] = True
         return self.setup_test_homeserver(config=config)
 
     def prepare(self, reactor, clock, hs):

tests/utils.py

@@ -168,9 +168,6 @@ def default_config(name, parse=False):
         # We need a sane default_room_version, otherwise attempts to create
         # rooms will fail.
         "default_room_version": DEFAULT_ROOM_VERSION,
-        # disable user directory updates, because they get done in the
-        # background, which upsets the test runner.
-        "update_user_directory": False,
         "caches": {"global_factor": 1},
         "listeners": [{"port": 0, "type": "http"}],
     }