diff --git a/synapse/handlers/room_member.py b/synapse/handlers/room_member.py
index 9e9f6cd062..ee2e807afc 100644
--- a/synapse/handlers/room_member.py
+++ b/synapse/handlers/room_member.py
@@ -51,6 +51,7 @@ from synapse.handlers.worker_lock import NEW_EVENT_DURING_PURGE_LOCK_NAME
from synapse.logging import opentracing
from synapse.metrics import event_processing_positions
from synapse.metrics.background_process_metrics import run_as_background_process
+from synapse.replication.http.push import ReplicationCopyPusherRestServlet
from synapse.storage.databases.main.state_deltas import StateDelta
from synapse.types import (
JsonDict,
@@ -181,6 +182,10 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
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._copy_push_client = ReplicationCopyPusherRestServlet.make_client(hs)
+
def _on_user_joined_room(self, event_id: str, room_id: str) -> None:
"""Notify the rate limiter that a room join has occurred.
@@ -1301,9 +1306,17 @@ class RoomMemberHandler(metaclass=abc.ABCMeta):
old_room_id, new_room_id, user_id
)
# Copy over push rules
- await self.store.copy_push_rules_from_room_to_room_for_user(
- old_room_id, new_room_id, user_id
- )
+ if self._is_push_writer:
+ await self.store.copy_push_rules_from_room_to_room_for_user(
+ old_room_id, new_room_id, user_id
+ )
+ else:
+ await self._copy_push_client(
+ instance_name=self._push_writer,
+ user_id=user_id,
+ old_room_id=old_room_id,
+ new_room_id=new_room_id,
+ )
except Exception:
logger.exception(
"Error copying tags and/or push rules from rooms %s to %s for user %s. "
|
