Allow background tasks to be run on a separate worker.

changelog.d/8369.feature

@@ -0,0 +1 @@
+Allow running background tasks in a separate worker process.

docs/sample_config.yaml

@@ -2458,6 +2458,12 @@ opentracing:
 #  events: worker1
 #  typing: worker1
 
+# The worker that is used to run background tasks (e.g. cleaning up expired
+# data). This should be one of the workers from `instance_map`. If not
+# provided this defaults to the main process.
+#
+#run_background_tasks: worker1
+
 
 # Configuration for Redis when using workers. This *must* be enabled when
 # using workers (unless using old style direct TCP configuration).

docs/workers.md

@@ -303,6 +303,23 @@ stream_writers:
     events: event_persister1
 ```
 
+#### Background tasks
+
+There is also *experimental* support for moving background tasks to a separate
+worker. Background tasks are run periodically or started via replication. Exactly
+which tasks are configured to run depends on your Synapse configuration (e.g. if
+stats is enabled).
+
+To enable this, the worker must have a `worker_name` and be listed in the
+`instance_map` config. For example, to move background tasks to a dedicated
+worker, the shared configuration would include:
+
+```yaml
+instance_map:
+    background_worker: null
+
+run_background_tasks: background_worker
+```
 
 ### `synapse.app.pusher`
 

synapse/app/generic_worker.py

@@ -133,6 +133,7 @@
 )
 from synapse.storage.databases.main.presence import UserPresenceState
 from synapse.storage.databases.main.search import SearchWorkerStore
+from synapse.storage.databases.main.stats import StatsStore
 from synapse.storage.databases.main.ui_auth import UIAuthWorkerStore
 from synapse.storage.databases.main.user_directory import UserDirectoryStore
 from synapse.types import ReadReceipt
@@ -454,6 +455,7 @@ class GenericWorkerSlavedStore(
     # FIXME(#3714): We need to add UserDirectoryStore as we write directly
     # rather than going via the correct worker.
     UserDirectoryStore,
+    StatsStore,
     UIAuthWorkerStore,
     SlavedDeviceInboxStore,
     SlavedDeviceStore,

synapse/app/homeserver.py

@@ -389,8 +389,6 @@ def setup(config_options):
     except UpgradeDatabaseException as e:
         quit_with_error("Failed to upgrade database: %s" % (e,))
 
-    hs.setup_master()
-
     async def do_acme() -> bool:
         """
         Reprovision an ACME certificate, if it's required.
@@ -620,16 +618,18 @@ def generate_user_daily_visit_stats():
     # Rather than update on per session basis, batch up the requests.
     # If you increase the loop period, the accuracy of user_daily_visits
     # table will decrease
-    clock.looping_call(generate_user_daily_visit_stats, 5 * 60 * 1000)
+    if hs.config.run_background_tasks:
+        clock.looping_call(generate_user_daily_visit_stats, 5 * 60 * 1000)
 
     # monthly active user limiting functionality
     def reap_monthly_active_users():
         return run_as_background_process(
             "reap_monthly_active_users", hs.get_datastore().reap_monthly_active_users
         )
 
-    clock.looping_call(reap_monthly_active_users, 1000 * 60 * 60)
-    reap_monthly_active_users()
+    if hs.config.run_background_tasks:
+        clock.looping_call(reap_monthly_active_users, 1000 * 60 * 60)
+        reap_monthly_active_users()
 
     async def generate_monthly_active_users():
         current_mau_count = 0
@@ -655,12 +655,14 @@ def start_generate_monthly_active_users():
             "generate_monthly_active_users", generate_monthly_active_users
         )
 
-    start_generate_monthly_active_users()
-    if hs.config.limit_usage_by_mau or hs.config.mau_stats_only:
+    if hs.config.run_background_tasks and (
+        hs.config.limit_usage_by_mau or hs.config.mau_stats_only
+    ):
+        start_generate_monthly_active_users()
         clock.looping_call(start_generate_monthly_active_users, 5 * 60 * 1000)
     # End of monthly active user settings
 
-    if hs.config.report_stats:
+    if hs.config.run_background_tasks and hs.config.report_stats:
         logger.info("Scheduling stats reporting for 3 hour intervals")
         clock.looping_call(start_phone_stats_home, 3 * 60 * 60 * 1000)
 

synapse/config/workers.py

@@ -109,10 +109,12 @@ def read_config(self, config, **kwargs):
             federation_sender_instances
         )
 
-        # A map from instance name to host/port of their HTTP replication endpoint.
+        # A map from instance name to host/port of their HTTP replication endpoint
+        # (or None if there's no HTTP replication endpoint).
         instance_map = config.get("instance_map") or {}
         self.instance_map = {
-            name: InstanceLocationConfig(**c) for name, c in instance_map.items()
+            name: InstanceLocationConfig(**c) if c else None
+            for name, c in instance_map.items()
         }
 
         # Map from type of streams to source, c.f. WriterLocations.
@@ -132,6 +134,24 @@ def read_config(self, config, **kwargs):
 
         self.events_shard_config = ShardedWorkerHandlingConfig(self.writers.events)
 
+        # Whether this worker should run background tasks or not.
+        #
+        # As a note for developers, the background tasks guarded by this should
+        # be able to run on only a single instance (meaning that they don't
+        # depend on any in-memory state of a particular worker).
+        #
+        # Effort is not made to ensure only a single instance of these tasks is
+        # running.
+        instance = config.get("run_background_tasks_on") or "master"
+        if instance != "master" and instance not in self.instance_map:
+            raise ConfigError(
+                "Instance %r is configured to run background tasks but does not appear in `instance_map` config."
+                % (instance,)
+            )
+        self.run_background_tasks = (
+            self.worker_name is None and instance == "master"
+        ) or self.worker_name == instance
+
     def generate_config_section(self, config_dir_path, server_name, **kwargs):
         return """\
         ## Workers ##
@@ -167,6 +187,12 @@ def generate_config_section(self, config_dir_path, server_name, **kwargs):
         #stream_writers:
         #  events: worker1
         #  typing: worker1
+
+        # The worker that is used to run background tasks (e.g. cleaning up expired
+        # data). This should be one of the workers from `instance_map`. If not
+        # provided this defaults to the main process.
+        #
+        #run_background_tasks: worker1
         """
 
     def read_arguments(self, args):

synapse/handlers/auth.py

@@ -203,7 +203,7 @@ def __init__(self, hs):
         self._clock = self.hs.get_clock()
 
         # Expire old UI auth sessions after a period of time.
-        if hs.config.worker_app is None:
+        if hs.config.run_background_tasks:
             self._clock.looping_call(
                 run_as_background_process,
                 5 * 60 * 1000,

synapse/handlers/stats.py

@@ -49,7 +49,7 @@ def __init__(self, hs):
         # Guard to ensure we only process deltas one at a time
         self._is_processing = False
 
-        if hs.config.stats_enabled:
+        if self.stats_enabled and hs.config.run_background_tasks:
             self.notifier.add_replication_callback(self.notify_new_event)
 
             # We kick this off so that we don't have to wait for a change before

synapse/handlers/user_directory.py

@@ -54,7 +54,7 @@ def __init__(self, hs):
         # Guard to ensure we only process deltas one at a time
         self._is_processing = False
 
-        if self.update_user_directory:
+        if hs.config.run_background_tasks and self.update_user_directory:
             self.notifier.add_replication_callback(self.notify_new_event)
 
             # We kick this off so that we don't have to wait for a change before

synapse/server.py

@@ -185,7 +185,11 @@ class HomeServer(metaclass=abc.ABCMeta):
             we are listening on to provide HTTP services.
     """
 
-    REQUIRED_ON_MASTER_STARTUP = ["user_directory_handler", "stats_handler"]
+    REQUIRED_ON_BACKGROUND_TASK_STARTUP = [
+        "auth",
+        "stats",
+        "user_directory",
+    ]
 
     # This is overridden in derived application classes
     # (such as synapse.app.homeserver.SynapseHomeServer) and gives the class to be
@@ -251,14 +255,17 @@ def setup(self) -> None:
         self.datastores = Databases(self.DATASTORE_CLASS, self)
         logger.info("Finished setting up.")
 
-    def setup_master(self) -> None:
+        if self.config.run_background_tasks:
+            self.setup_background_tasks()
+
+    def setup_background_tasks(self) -> None:
         """
         Some handlers have side effects on instantiation (like registering
         background updates). This function causes them to be fetched, and
         therefore instantiated, to run those side effects.
         """
-        for i in self.REQUIRED_ON_MASTER_STARTUP:
-            getattr(self, "get_" + i)()
+        for i in self.REQUIRED_ON_BACKGROUND_TASK_STARTUP:
+            getattr(self, "get_" + i + "_handler")()
 
     def get_reactor(self) -> twisted.internet.base.ReactorBase:
         """

synapse/storage/databases/main/ui_auth.py

@@ -288,8 +288,6 @@ async def get_user_agents_ips_to_ui_auth_session(
         )
         return [(row["user_agent"], row["ip"]) for row in rows]
 
-
-class UIAuthStore(UIAuthWorkerStore):
     async def delete_old_ui_auth_sessions(self, expiration_time: int) -> None:
         """
         Remove sessions which were last used earlier than the expiration time.
@@ -339,3 +337,7 @@ def _delete_old_ui_auth_sessions_txn(
             iterable=session_ids,
             keyvalues={},
         )
+
+
+class UIAuthStore(UIAuthWorkerStore):
+    pass

tests/utils.py

@@ -276,7 +276,7 @@ def setup_test_homeserver(
 
         hs.setup()
         if homeserverToUse.__name__ == "TestHomeServer":
-            hs.setup_master()
+            hs.setup_background_tasks()
 
         if isinstance(db_engine, PostgresEngine):
             database = hs.get_datastores().databases[0]