Redirect event persistence traffic to a particular event persister worker

synapse/config/_base.py

@@ -947,6 +947,10 @@ class ShardedWorkerHandlingConfig:
 
     instances: List[str]
 
+    # A map of key to instance name. If any of these keys are used,
+    # the associated instance is *always* returned.
+    instances_reserved_for_keys: Dict[str, str] = {}
+
     def should_handle(self, instance_name: str, key: str) -> bool:
         """Whether this instance is responsible for handling the given key."""
         # If no instances are defined we assume some other worker is handling
@@ -964,6 +968,10 @@ class ShardedWorkerHandlingConfig:
         method by default.
         """
 
+        reserved_instance = self.instances_reserved_for_keys.get(key)
+        if reserved_instance is not None:
+            return reserved_instance
+
         if not self.instances:
             raise Exception("Unknown worker")
 

synapse/config/_base.pyi

@@ -199,7 +199,10 @@ def find_config_files(search_paths: List[str]) -> List[str]: ...
 
 class ShardedWorkerHandlingConfig:
     instances: List[str]
-    def __init__(self, instances: List[str]) -> None: ...
+    instances_reserved_for_keys: Dict[str, str]
+    def __init__(
+        self, instances: List[str], instances_reserved_for_keys: Dict[str, str] = {}
+    ) -> None: ...
     def should_handle(self, instance_name: str, key: str) -> bool: ...  # noqa: F811
 
 class RoutableShardedWorkerHandlingConfig(ShardedWorkerHandlingConfig):

synapse/config/workers.py

@@ -272,8 +272,19 @@ class WorkerConfig(Config):
                 "Must only specify one instance to handle `presence` messages."
             )
 
+        # Make the event shard config point specific rooms to a specific worker.
+        # All other rooms should have events persisted by a different worker.
+        chosen_worker_instance = "event_persister-4"
+        event_instances = self.writers.events.copy()
+        event_instances.remove(chosen_worker_instance)
+
+        instances_reserved_for_keys = {
+            "!ioWEdTBHIhOGYVKWyq:libera.chat": chosen_worker_instance,
+            "!bBgnAGciIvrtPXkHkp:libera.chat": chosen_worker_instance,
+        }
+
         self.events_shard_config = RoutableShardedWorkerHandlingConfig(
-            self.writers.events
+            event_instances, instances_reserved_for_keys
         )
 
         # Handle sharded push