Introduce update_user_directory_on

changelog.d/11450.feature

@@ -0,0 +1 @@
+Introduce `update_user_directory_on` config variable.

docker/configure_workers_and_start.py

@@ -50,7 +50,7 @@
         "endpoint_patterns": [
             "^/_matrix/client/(api/v1|r0|v3|unstable)/user_directory/search$"
         ],
-        "shared_extra_conf": {"update_user_directory": False},
+        "shared_extra_conf": {"update_user_directory_on": "user_dir1"},
         "worker_extra_conf": "",
     },
     "media_repository": {

docs/sample_config.yaml

@@ -2657,6 +2657,13 @@ opentracing:
 #
 #run_background_tasks_on: worker1
 
+# The worker that is used to run user directory update tasks
+# (e.g. users in public rooms, shared rooms between users.).
+#
+# If not provided, or null, this defaults to the main process.
+#
+#update_user_directory_on: "sync1"
+
 # A shared secret used by the replication APIs to authenticate HTTP requests
 # from workers.
 #

docs/workers.md

@@ -374,7 +374,7 @@ the shared configuration would include:
 run_background_tasks_on: background_worker
 ```
 
-You might also wish to investigate the `update_user_directory` and
+You might also wish to investigate the `update_user_directory_on` and
 `media_instance_running_background_jobs` settings.
 
 ### `synapse.app.pusher`
@@ -467,9 +467,10 @@ the following regular expressions:
 
     ^/_matrix/client/(api/v1|r0|v3|unstable)/user_directory/search$
 
-When using this worker you must also set `update_user_directory: False` in the
-shared configuration file to stop the main synapse running background
-jobs related to updating the user directory.
+When using this worker you must also set `update_user_directory_on` in the
+shared configuration file to the user directory worker's name to stop the main 
+synapse running background jobs related to updating the user directory, and give
+the user directory worker the exclusive right to update it instead.
 
 ### `synapse.app.frontend_proxy`
 

synapse/app/admin_cmd.py

@@ -211,7 +211,7 @@ def start(config_options: List[str]) -> None:
         config.logging.no_redirect_stdio = True
 
     # Explicitly disable background processes
-    config.server.update_user_directory = False
+    config.worker.update_user_directory = False
     config.worker.run_background_tasks = False
     config.worker.start_pushers = False
     config.worker.pusher_shard_config.instances = []

synapse/app/generic_worker.py

@@ -40,7 +40,7 @@
 from synapse.config._base import ConfigError
 from synapse.config.homeserver import HomeServerConfig
 from synapse.config.logger import setup_logging
-from synapse.config.server import ListenerConfig
+from synapse.config.workers import ListenerConfig
 from synapse.federation.transport.server import TransportLayerServer
 from synapse.http.server import JsonResource, OptionsResource
 from synapse.http.servlet import RestServlet, parse_json_object_from_request
@@ -126,6 +126,11 @@
 
 logger = logging.getLogger("synapse.app.generic_worker")
 
+DIRECTORY_UPDATE_WARNING = """
+The user directory worker is not allowed to perform user directory updates per the config.
+Please add or edit ``update_user_directory_on: "{0}"`` in the config'
+"""
+
 
 class KeyUploadServlet(RestServlet):
     """An implementation of the `KeyUploadServlet` that responds to read only
@@ -458,20 +463,8 @@ def start(config_options: List[str]) -> None:
         config.appservice.notify_appservices = False
 
     if config.worker.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
+        if not config.worker.update_user_directory:
+            logger.warning(DIRECTORY_UPDATE_WARNING.format(config.worker.worker_name))
 
     synapse.events.USE_FROZEN_DICTS = config.server.use_frozen_dicts
     synapse.util.caches.TRACK_MEMORY_USAGE = config.caches.track_memory_usage

synapse/config/server.py

@@ -323,10 +323,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

@@ -14,6 +14,7 @@
 # limitations under the License.
 
 import argparse
+import logging
 from typing import List, Union
 
 import attr
@@ -26,6 +27,8 @@
 )
 from .server import ListenerConfig, parse_listener_def
 
+logger = logging.Logger(__name__)
+
 _FEDERATION_SENDER_WITH_SEND_FEDERATION_ENABLED_ERROR = """
 The send_federation config option must be disabled in the main
 synapse process before they can be run in a separate worker.
@@ -40,6 +43,12 @@
 Please add ``start_pushers: false`` to the main config
 """
 
+USER_UPDATE_DIRECTORY_DEPRECATION_WARNING = """
+The 'update_user_directory' configuration setting is now deprecated,
+and replaced with 'update_user_directory_on'.  See the sample configuration
+file for details of 'update_user_directory_on'.
+"""
+
 
 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
@@ -292,9 +301,36 @@ def read_config(self, config, **kwargs):
         # No effort is made to ensure only a single instance of these tasks is
         # 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
+        self.run_background_tasks = self.instance_name == background_tasks_instance
+
+        # update_user_directory_on controls which worker is responsible for updating the user directory.
+        # None means that the main process should handle this, as does the absence of the config key.
+        #
+        # Note that due to backwards compatibility, we must also test for update_user_directory, and apply it if
+        # update_user_directory_on is not defined at the same time.
+
+        if "update_user_directory" in config:
+            # Backwards compatibility case
+            logger.warning(USER_UPDATE_DIRECTORY_DEPRECATION_WARNING)
+
+            if config["update_user_directory"]:
+                # Main process should update
+                update_user_directory_on = "master"
+            else:
+                # user directory worker should update
+                update_user_directory_on = (
+                    self.instance_name
+                    if self.worker_app == "synapse.app.user_dir"
+                    else ""
+                )
+        else:
+            update_user_directory_on = (
+                config.get("update_user_directory_on") or "master"
+            )
+
+        self.update_user_directory: bool = (
+            self.instance_name == update_user_directory_on
+        )
 
     def generate_config_section(self, config_dir_path, server_name, **kwargs):
         return """\
@@ -337,6 +373,13 @@ def generate_config_section(self, config_dir_path, server_name, **kwargs):
         #
         #run_background_tasks_on: worker1
 
+        # The worker that is used to run user directory update tasks
+        # (e.g. users in public rooms, shared rooms between users.).
+        #
+        # If not provided, or null, this defaults to the main process.
+        #
+        #update_user_directory_on: "sync1"
+
         # A shared secret used by the replication APIs to authenticate HTTP requests
         # from workers.
         #

synapse/handlers/user_directory.py

@@ -60,7 +60,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.server.update_user_directory
+        self.update_user_directory = hs.config.worker.update_user_directory
         self.search_all_users = hs.config.userdirectory.user_directory_search_all_users
         self.spam_checker = hs.get_spam_checker()
         # The current position in the current_state_delta stream

synapse/rest/client/shared_rooms.py

@@ -42,19 +42,11 @@ def __init__(self, hs: "HomeServer"):
         super().__init__()
         self.auth = hs.get_auth()
         self.store = hs.get_datastore()
-        self.user_directory_active = hs.config.server.update_user_directory
 
     async def on_GET(
         self, request: SynapseRequest, user_id: str
     ) -> Tuple[int, JsonDict]:
 
-        if not self.user_directory_active:
-            raise SynapseError(
-                code=400,
-                msg="The user directory is disabled on this server. Cannot determine shared rooms.",
-                errcode=Codes.FORBIDDEN,
-            )
-
         UserID.from_string(user_id)
 
         requester = await self.auth.get_user_by_req(request)

tests/handlers/test_user_directory.py

@@ -56,7 +56,8 @@ class UserDirectoryTestCase(unittest.HomeserverTestCase):
 
     def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
         config = self.default_config()
-        config["update_user_directory"] = True
+        # Always process user directory updates on this main homeserver process.
+        config["update_user_directory_on"] = None
 
         self.appservice = ApplicationService(
             token="i_am_an_app_service",
@@ -1014,7 +1015,7 @@ class TestUserDirSearchDisabled(unittest.HomeserverTestCase):
 
     def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
         config = self.default_config()
-        config["update_user_directory"] = True
+        config["update_user_directory_on"] = None
         hs = self.setup_test_homeserver(config=config)
 
         self.config = hs.config

tests/rest/client/test_shared_rooms.py

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

tests/utils.py

@@ -169,7 +169,7 @@ def default_config(name, parse=False):
         "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,
+        "update_user_directory_on": "(non-existent worker)",
         "caches": {"global_factor": 1},
         "listeners": [{"port": 0, "type": "http"}],
     }