Allow excluding specific rooms from presence updates fan-out

changelog.d/19920.feature

@@ -0,0 +1 @@
+Add a new `exclude_rooms_from_presence` config option to stop presence being routed between users solely because they share one of the listed rooms, avoiding presence fan-out in large hidden "directory" rooms.

docs/usage/configuration/config_documentation.md

@@ -4292,6 +4292,18 @@ exclude_rooms_from_sync:
 - '!foo:example.com'
 ```
 ---
+### `exclude_rooms_from_presence`
+
+*(array)* A list of room IDs whose membership should not, on its own, cause presence updates to be routed between their members. A user will still exchange presence with anyone they share a *non-excluded* room with, and will always receive their own presence.
+
+This is useful alongside large auto-joined "directory" rooms (e.g. rooms created via `auto_join_rooms` and hidden with [`exclude_rooms_from_sync`](#exclude_rooms_from_sync)): without this option, every presence change from any member of such a room fans out to every other member, which can overwhelm the presence writer. Defaults to `[]`.
+
+Example configuration:
+```yaml
+exclude_rooms_from_presence:
+- '!foo:example.com'
+```
+---
 ## Opentracing
 
 Configuration options related to Opentracing support.

synapse/config/server.py

@@ -896,6 +896,10 @@ def read_config(self, config: JsonDict, **kwargs: Any) -> None:
             config.get("exclude_rooms_from_sync") or []
         )
 
+        self.rooms_to_exclude_from_presence: list[str] = (
+            config.get("exclude_rooms_from_presence") or []
+        )
+
         delete_stale_devices_after: str | None = (
             config.get("delete_stale_devices_after") or None
         )

synapse/handlers/presence.py

@@ -225,6 +225,12 @@ def __init__(self, hs: "HomeServer"):
         self._presence_enabled = hs.config.server.presence_enabled
         self._track_presence = hs.config.server.track_presence
 
+        # Rooms which, on their own, should not cause presence to be routed
+        # between their members. See `exclude_rooms_from_presence` in the config.
+        self._rooms_to_exclude_from_presence = frozenset(
+            hs.config.server.rooms_to_exclude_from_presence
+        )
+
         self._federation = None
         if hs.should_send_federation():
             self._federation = hs.get_federation_sender()
@@ -431,6 +437,7 @@ async def maybe_send_presence_to_interested_destinations(
             self.store,
             self.presence_router,
             states,
+            self._rooms_to_exclude_from_presence,
         )
 
         for destinations, host_states in hosts_to_states:
@@ -640,7 +647,12 @@ def _user_syncing() -> Generator[None, None, None]:
     async def notify_from_replication(
         self, states: list[UserPresenceState], stream_id: int
     ) -> None:
-        parties = await get_interested_parties(self.store, self.presence_router, states)
+        parties = await get_interested_parties(
+            self.store,
+            self.presence_router,
+            states,
+            self._rooms_to_exclude_from_presence,
+        )
         room_ids_to_states, users_to_states = parties
 
         self.notifier.on_new_event(
@@ -1044,6 +1056,7 @@ async def _update_states(
                     self.store,
                     self.presence_router,
                     list(to_federation_ping.values()),
+                    self._rooms_to_exclude_from_presence,
                 )
 
                 for destinations, states in hosts_to_states:
@@ -1314,7 +1327,12 @@ async def _persist_and_notify(self, states: list[UserPresenceState]) -> None:
         """
         stream_id, max_token = await self.store.update_presence(states)
 
-        parties = await get_interested_parties(self.store, self.presence_router, states)
+        parties = await get_interested_parties(
+            self.store,
+            self.presence_router,
+            states,
+            self._rooms_to_exclude_from_presence,
+        )
         room_ids_to_states, users_to_states = parties
 
         self.notifier.on_new_event(
@@ -1461,7 +1479,10 @@ async def is_visible(self, observed_user: UserID, observer_user: UserID) -> bool
             observed_user.to_string()
         )
 
-        if observer_room_ids & observed_room_ids:
+        shared_room_ids = (
+            observer_room_ids & observed_room_ids
+        ) - self._rooms_to_exclude_from_presence
+        if shared_room_ids:
             return True
 
         return False
@@ -1572,6 +1593,14 @@ async def _handle_state_delta(self, room_id: str, deltas: list[StateDelta]) -> N
         to be handled.
         """
 
+        # Rooms on the presence exclusion list should not, on their own, cause
+        # presence to be shared between their members. Everything this method
+        # does is per-room presence fan-out (enumerating the room's members and
+        # sending their presence to local users / remote hosts), so we can skip
+        # excluded rooms entirely and avoid the expensive enumeration.
+        if room_id in self._rooms_to_exclude_from_presence:
+            return
+
         # Sets of newly joined users. Note that if the local server is
         # joining a remote room for the first time we'll see both the joining
         # user and all remote users as newly joined.
@@ -1827,6 +1856,9 @@ def __init__(self, hs: "HomeServer"):
         self.server_name = hs.hostname
         self.clock = hs.get_clock()
         self.store = hs.get_datastores().main
+        self._rooms_to_exclude_from_presence = frozenset(
+            hs.config.server.rooms_to_exclude_from_presence
+        )
 
     async def get_new_events(
         self,
@@ -1942,7 +1974,9 @@ async def get_new_events(
                     ).inc()
 
                     sharing_users = await self.store.do_users_share_a_room(
-                        user_id, updated_users
+                        user_id,
+                        updated_users,
+                        self._rooms_to_exclude_from_presence,
                     )
 
                     interested_and_updated_users = (
@@ -1958,7 +1992,9 @@ async def get_new_events(
                     ).inc()
 
                     users_interested_in = (
-                        await self.store.get_users_who_share_room_with_user(user_id)
+                        await self.store.get_users_who_share_room_with_user(
+                            user_id, self._rooms_to_exclude_from_presence
+                        )
                     )
                     users_interested_in.update(additional_users_interested_in)
 
@@ -1971,7 +2007,9 @@ async def get_new_events(
                 # No from_key has been specified. Return the presence for all users
                 # this user is interested in
                 interested_and_updated_users = (
-                    await self.store.get_users_who_share_room_with_user(user_id)
+                    await self.store.get_users_who_share_room_with_user(
+                        user_id, self._rooms_to_exclude_from_presence
+                    )
                 )
                 interested_and_updated_users.update(additional_users_interested_in)
 
@@ -2335,7 +2373,10 @@ def _combine_device_states(
 
 
 async def get_interested_parties(
-    store: DataStore, presence_router: PresenceRouter, states: list[UserPresenceState]
+    store: DataStore,
+    presence_router: PresenceRouter,
+    states: list[UserPresenceState],
+    excluded_rooms: AbstractSet[str] = frozenset(),
 ) -> tuple[dict[str, list[UserPresenceState]], dict[str, list[UserPresenceState]]]:
     """Given a list of states return which entities (rooms, users)
     are interested in the given states.
@@ -2344,6 +2385,8 @@ async def get_interested_parties(
         store: The homeserver's data store.
         presence_router: A module for augmenting the destinations for presence updates.
         states: A list of incoming user presence updates.
+        excluded_rooms: Rooms which should not, on their own, cause presence to
+            be routed between their members.
 
     Returns:
         A 2-tuple of `(room_ids_to_states, users_to_states)`,
@@ -2354,6 +2397,8 @@ async def get_interested_parties(
     for state in states:
         room_ids = await store.get_rooms_for_user(state.user_id)
         for room_id in room_ids:
+            if room_id in excluded_rooms:
+                continue
             room_ids_to_states.setdefault(room_id, []).append(state)
 
         # Always notify self
@@ -2374,6 +2419,7 @@ async def get_interested_remotes(
     store: DataStore,
     presence_router: PresenceRouter,
     states: list[UserPresenceState],
+    excluded_rooms: AbstractSet[str] = frozenset(),
 ) -> list[tuple[StrCollection, Collection[UserPresenceState]]]:
     """Given a list of presence states figure out which remote servers
     should be sent which.
@@ -2384,6 +2430,8 @@ async def get_interested_remotes(
         store: The homeserver's data store.
         presence_router: A module for augmenting the destinations for presence updates.
         states: A list of incoming user presence updates.
+        excluded_rooms: Rooms which should not, on their own, cause presence to
+            be routed to their remote members.
 
     Returns:
         A map from destinations to presence states to send to that destination.
@@ -2397,6 +2445,8 @@ async def get_interested_remotes(
         room_ids = await store.get_rooms_for_user(state.user_id)
         hosts: set[str] = set()
         for room_id in room_ids:
+            if room_id in excluded_rooms:
+                continue
             room_hosts = await store.get_current_hosts_in_room(room_id)
             hosts.update(room_hosts)
         hosts_and_states.append((hosts, [state]))

synapse/storage/databases/main/roommember.py

@@ -844,21 +844,30 @@ async def get_rooms_for_users(
 
     @cached(max_entries=10000)
     async def does_pair_of_users_share_a_room(
-        self, user_id: str, other_user_id: str
+        self, user_id: str, other_user_id: str, excluded_rooms: AbstractSet[str]
     ) -> bool:
         raise NotImplementedError()
 
     @cachedList(
         cached_method_name="does_pair_of_users_share_a_room", list_name="other_user_ids"
     )
     async def _do_users_share_a_room(
-        self, user_id: str, other_user_ids: Collection[str]
+        self,
+        user_id: str,
+        other_user_ids: Collection[str],
+        excluded_rooms: AbstractSet[str],
     ) -> Mapping[str, bool | None]:
         """Return mapping from user ID to whether they share a room with the
         given user.
 
         Note: `None` and `False` are equivalent and mean they don't share a
         room.
+
+        Args:
+            user_id: The user to check for shared rooms against.
+            other_user_ids: The users to check share a room with `user_id`.
+            excluded_rooms: Rooms which should not, on their own, count as a
+                shared room.
         """
 
         def do_users_share_a_room_txn(
@@ -868,22 +877,34 @@ def do_users_share_a_room_txn(
                 self.database_engine, "state_key", user_ids
             )
 
+            # We optionally exclude some of the target user's rooms, so that two
+            # users who only share an excluded room aren't considered to share a
+            # room at all.
+            if excluded_rooms:
+                excluded_clause, excluded_args = make_in_list_sql_clause(
+                    self.database_engine, "room_id", excluded_rooms, negative=True
+                )
+                excluded_clause = "AND " + excluded_clause
+            else:
+                excluded_clause = ""
+                excluded_args = []
+
             # This query works by fetching both the list of rooms for the target
             # user and the set of other users, and then checking if there is any
             # overlap.
             sql = f"""
                 SELECT DISTINCT b.state_key
                 FROM (
                     SELECT room_id FROM current_state_events
-                    WHERE type = 'm.room.member' AND membership = 'join' AND state_key = ?
+                    WHERE type = 'm.room.member' AND membership = 'join' AND state_key = ? {excluded_clause}
                 ) AS a
                 INNER JOIN (
                     SELECT room_id, state_key FROM current_state_events
                     WHERE type = 'm.room.member' AND membership = 'join' AND {clause}
                 ) AS b using (room_id)
             """
 
-            txn.execute(sql, (user_id, *args))
+            txn.execute(sql, (user_id, *excluded_args, *args))
             return {u: True for (u,) in txn}
 
         to_return = {}
@@ -896,11 +917,23 @@ def do_users_share_a_room_txn(
         return to_return
 
     async def do_users_share_a_room(
-        self, user_id: str, other_user_ids: Collection[str]
+        self,
+        user_id: str,
+        other_user_ids: Collection[str],
+        excluded_rooms: AbstractSet[str] = frozenset(),
     ) -> set[str]:
-        """Return the set of users who share a room with the first users"""
+        """Return the set of users who share a room with the first users.
+
+        Args:
+            user_id: The user to check for shared rooms against.
+            other_user_ids: The users to check share a room with `user_id`.
+            excluded_rooms: Rooms which should not, on their own, count as a
+                shared room.
+        """
 
-        user_dict = await self._do_users_share_a_room(user_id, other_user_ids)
+        user_dict = await self._do_users_share_a_room(
+            user_id, other_user_ids, excluded_rooms
+        )
 
         return {u for u, share_room in user_dict.items() if share_room}
 
@@ -971,12 +1004,22 @@ async def do_users_share_a_room_joined_or_invited(
 
         return {u for u, share_room in user_dict.items() if share_room}
 
-    async def get_users_who_share_room_with_user(self, user_id: str) -> set[str]:
-        """Returns the set of users who share a room with `user_id`"""
+    async def get_users_who_share_room_with_user(
+        self, user_id: str, excluded_rooms: AbstractSet[str] = frozenset()
+    ) -> set[str]:
+        """Returns the set of users who share a room with `user_id`.
+
+        Args:
+            user_id: The user to find the co-occupants of.
+            excluded_rooms: Rooms which should not, on their own, count as a
+                shared room.
+        """
         room_ids = await self.get_rooms_for_user(user_id)
 
         user_who_share_room: set[str] = set()
         for room_id in room_ids:
+            if room_id in excluded_rooms:
+                continue
             user_ids = await self.get_users_in_room(room_id)
             user_who_share_room.update(user_ids)