Allow background tasks to be run on a separate worker.

Author: clokep

changelog.d/8369.feature

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

synapse/app/generic_worker.py

@@ -885,6 +885,7 @@ def start(config_options):
     # For backwards compatibility let any of the old app names.
     assert config.worker_app in (
         "synapse.app.appservice",
+        "synapse.app.background_worker",
         "synapse.app.client_reader",
         "synapse.app.event_creator",
         "synapse.app.federation_reader",
@@ -961,6 +962,22 @@ def start(config_options):
         # For other worker types we force this to off.
         config.worker.send_federation = False
 
+    if config.worker_app == "synapse.app.background_worker":
+        if config.worker.run_background_tasks:
+            sys.stderr.write(
+                "\nThe run_background_tasks must be disabled in the main synapse process"
+                "\nbefore they can be run in a separate worker."
+                "\nPlease add ``run_background_tasks: false`` to the main config"
+                "\n"
+            )
+            sys.exit(1)
+
+            # Force the background tasks to start since they will be disabled in the main config
+            config.worker.run_background_tasks = True
+        else:
+            # For other worker types we force this to off.
+            config.worker.run_background_tasks = False
+
     synapse.events.USE_FROZEN_DICTS = config.use_frozen_dicts
 
     hs = GenericWorkerServer(

synapse/app/homeserver.py

@@ -620,16 +620,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
@@ -656,11 +658,13 @@ def start_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
+    ):
         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

@@ -132,6 +132,16 @@ 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.
+        self.run_background_tasks = config.get("run_background_tasks", True)
+
     def generate_config_section(self, config_dir_path, server_name, **kwargs):
         return """\
         ## Workers ##

synapse/handlers/account_validity.py

@@ -42,6 +42,7 @@ def __init__(self, hs):
         if (
             self._account_validity.enabled
             and self._account_validity.renew_by_email_enabled
+            and hs.config.run_background_tasks
         ):
             # Don't do email-specific configuration if renewal by email is disabled.
             self._template_html = self.config.account_validity_template_html

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/deactivate_account.py

@@ -42,7 +42,7 @@ def __init__(self, hs):
 
         # Start the user parter loop so it can resume parting users from rooms where
         # it left off (if it has work left to do).
-        if hs.config.worker_app is None:
+        if hs.config.run_background_tasks:
             hs.get_reactor().callWhenRunning(self._start_user_parting)
 
         self._account_validity_enabled = hs.config.account_validity.enabled

synapse/handlers/device.py

@@ -522,12 +522,13 @@ def __init__(self, hs, device_handler):
 
         # Attempt to resync out of sync device lists every 30s.
         self._resync_retry_in_progress = False
-        self.clock.looping_call(
-            run_as_background_process,
-            30 * 1000,
-            func=self._maybe_retry_device_resync,
-            desc="_maybe_retry_device_resync",
-        )
+        if hs.config.run_background_tasks:
+            self.clock.looping_call(
+                run_as_background_process,
+                30 * 1000,
+                func=self._maybe_retry_device_resync,
+                desc="_maybe_retry_device_resync",
+            )
 
     @trace
     async def incoming_device_list_update(self, origin, edu_content):

synapse/handlers/message.py

@@ -413,7 +413,7 @@ def __init__(self, hs: "HomeServer"):
             self._consent_uri_builder = ConsentURIBuilder(self.config)
 
         if (
-            not self.config.worker_app
+            self.config.run_background_tasks
             and self.config.cleanup_extremities_with_dummy_events
         ):
             self.clock.looping_call(

synapse/handlers/pagination.py

@@ -92,7 +92,7 @@ def __init__(self, hs: "HomeServer"):
         self._retention_allowed_lifetime_min = hs.config.retention_allowed_lifetime_min
         self._retention_allowed_lifetime_max = hs.config.retention_allowed_lifetime_max
 
-        if hs.config.retention_enabled:
+        if hs.config.run_background_tasks and hs.config.retention_enabled:
             # Run the purge jobs described in the configuration file.
             for job in hs.config.retention_purge_jobs:
                 logger.info("Setting up purge job with config: %s", job)