Fixups to new push stream (#17038)

Follow on from #17037

changelog.d/17037.feature

@@ -1 +1 @@
-Add support for moving `/push_rules` off of main process.
+Add support for moving `/pushrules` off of main process.

changelog.d/17038.feature

@@ -0,0 +1 @@
+Add support for moving `/pushrules` off of main process.

docker/configure_workers_and_start.py

@@ -310,6 +310,13 @@
         "shared_extra_conf": {},
         "worker_extra_conf": "",
     },
+    "push_rules": {
+        "app": "synapse.app.generic_worker",
+        "listener_resources": ["client", "replication"],
+        "endpoint_patterns": ["^/_matrix/client/(api/v1|r0|v3|unstable)/pushrules/"],
+        "shared_extra_conf": {},
+        "worker_extra_conf": "",
+    },
 }
 
 # Templates for sections that may be inserted multiple times in config files
@@ -401,6 +408,7 @@ def add_worker_roles_to_shared_config(
         "receipts",
         "to_device",
         "typing",
+        "push_rules",
     ]
 
     # Worker-type specific sharding config. Now a single worker can fulfill multiple

docs/workers.md

@@ -532,12 +532,12 @@ the stream writer for the `presence` stream:
 
     ^/_matrix/client/(api/v1|r0|v3|unstable)/presence/
 
-##### The `push` stream
+##### The `push_rules` stream
 
 The following endpoints should be routed directly to the worker configured as
 the stream writer for the `push` stream:
 
-    ^/_matrix/client/(api/v1|r0|v3|unstable)/push_rules/
+    ^/_matrix/client/(api/v1|r0|v3|unstable)/pushrules/
 
 #### Restrict outbound federation traffic to a specific set of workers
 

synapse/config/workers.py

@@ -156,7 +156,7 @@ class WriterLocations:
             can only be a single instance.
         presence: The instances that write to the presence stream. Currently
             can only be a single instance.
-        push: The instances that write to the push stream. Currently
+        push_rules: The instances that write to the push stream. Currently
             can only be a single instance.
     """
 
@@ -184,7 +184,7 @@ class WriterLocations:
         default=["master"],
         converter=_instance_to_list_converter,
     )
-    push: List[str] = attr.ib(
+    push_rules: List[str] = attr.ib(
         default=["master"],
         converter=_instance_to_list_converter,
     )
@@ -347,7 +347,7 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None:
             "account_data",
             "receipts",
             "presence",
-            "push",
+            "push_rules",
         ):
             instances = _instance_to_list_converter(getattr(self.writers, stream))
             for instance in instances:
@@ -385,7 +385,7 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None:
                 "Must only specify one instance to handle `presence` messages."
             )
 
-        if len(self.writers.push) != 1:
+        if len(self.writers.push_rules) != 1:
             raise ConfigError(
                 "Must only specify one instance to handle `push` messages."
             )

synapse/handlers/room_member.py

@@ -182,8 +182,10 @@ def __init__(self, hs: "HomeServer"):
             hs.config.server.forgotten_room_retention_period
         )
 
-        self._is_push_writer = hs.get_instance_name() in hs.config.worker.writers.push
-        self._push_writer = hs.config.worker.writers.push[0]
+        self._is_push_writer = (
+            hs.get_instance_name() in hs.config.worker.writers.push_rules
+        )
+        self._push_writer = hs.config.worker.writers.push_rules[0]
         self._copy_push_client = ReplicationCopyPusherRestServlet.make_client(hs)
 
     def _on_user_joined_room(self, event_id: str, room_id: str) -> None:

synapse/replication/tcp/handler.py

@@ -180,7 +180,7 @@ def __init__(self, hs: "HomeServer"):
                 continue
 
             if isinstance(stream, PushRulesStream):
-                if hs.get_instance_name() in hs.config.worker.writers.push:
+                if hs.get_instance_name() in hs.config.worker.writers.push_rules:
                     self._streams_to_replicate.append(stream)
 
                 continue

synapse/rest/client/push_rule.py

@@ -59,7 +59,9 @@ def __init__(self, hs: "HomeServer"):
         self.auth = hs.get_auth()
         self.store = hs.get_datastores().main
         self.notifier = hs.get_notifier()
-        self._is_push_worker = hs.get_instance_name() in hs.config.worker.writers.push
+        self._is_push_worker = (
+            hs.get_instance_name() in hs.config.worker.writers.push_rules
+        )
         self._push_rules_handler = hs.get_push_rules_handler()
         self._push_rule_linearizer = Linearizer(name="push_rules")
 

synapse/storage/databases/main/push_rule.py

@@ -136,7 +136,9 @@ def __init__(
     ):
         super().__init__(database, db_conn, hs)
 
-        self._is_push_writer = hs.get_instance_name() in hs.config.worker.writers.push
+        self._is_push_writer = (
+            hs.get_instance_name() in hs.config.worker.writers.push_rules
+        )
 
         # In the worker store this is an ID tracker which we overwrite in the non-worker
         # class below that is used on the main process.