Improve caching for presence (#19939)

This does two things, first it adds a config flag to ignore rooms for
the purposes of presence routing.

Secondly, it changes the caching behaviour to try and improve the cache
hit ratio. Previously, the size of the `do_users_share_a_room` cache
(which stores pairs of users) needs to `O(n²)` for the number of online
users, which is infeasible for large servers.

Instead, we call `get_users_in_room` for both the syncing and updated
users. This sounds more expensive, but a) we will already have cached
the syncing user's rooms, and b) we will only calculate the updated
user's rooms once (rather than once per syncing user).

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>

changelog.d/19935.feature

@@ -0,0 +1 @@
+Add an `exclude_rooms_from_presence` configuration option to stop presence being routed between users solely because they share one of the listed rooms.

changelog.d/19939.misc

@@ -0,0 +1 @@
+Minor presence performance improvements for large servers.

docs/usage/configuration/config_documentation.md

@@ -4310,6 +4310,16 @@ exclude_rooms_from_sync:
 - '!foo:example.com'
 ```
 ---
+### `exclude_rooms_from_presence`
+
+*(array)* A list of rooms to exclude from presence updates. Presence will not be routed between two users solely because they share one of these rooms. Users who also share a non-excluded room continue to exchange presence as normal. Defaults to `[]`.
+
+Example configuration:
+```yaml
+exclude_rooms_from_presence:
+- '!foo:example.com'
+```
+---
 ## Opentracing
 
 Configuration options related to Opentracing support.

schema/synapse-config.schema.yaml

@@ -5354,6 +5354,18 @@ properties:
     default: []
     examples:
       - - "!foo:example.com"
+  exclude_rooms_from_presence:
+    type: array
+    description: >-
+      A list of rooms to exclude from presence updates. Presence will not be
+      routed between two users solely because they share one of these rooms.
+      Users who also share a non-excluded room continue to exchange presence as
+      normal.
+    items:
+      type: string
+    default: []
+    examples:
+      - - "!foo:example.com"
   opentracing:
     type: object
     description: >-

synapse/config/server.py

@@ -936,6 +936,10 @@ class ServerConfig(Config):
             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

@@ -222,6 +222,12 @@ class BasePresenceHandler(abc.ABC):
         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
+        )
+
         # The (configurable) presence state machine timers.
         self._last_active_granularity = (
             hs.config.server.presence_last_active_granularity
@@ -435,6 +441,7 @@ class BasePresenceHandler(abc.ABC):
             self.store,
             self.presence_router,
             states,
+            self._rooms_to_exclude_from_presence,
         )
 
         for destinations, host_states in hosts_to_states:
@@ -688,7 +695,12 @@ class WorkerPresenceHandler(BasePresenceHandler):
     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(
@@ -1141,6 +1153,7 @@ class PresenceHandler(BasePresenceHandler):
                     self.store,
                     self.presence_router,
                     list(to_federation_ping.values()),
+                    self._rooms_to_exclude_from_presence,
                 )
 
                 for destinations, states in hosts_to_states:
@@ -1414,7 +1427,12 @@ class PresenceHandler(BasePresenceHandler):
         """
         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(
@@ -1561,7 +1579,10 @@ class PresenceHandler(BasePresenceHandler):
             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
@@ -1672,6 +1693,12 @@ class PresenceHandler(BasePresenceHandler):
         to be handled.
         """
 
+        # Excluded rooms should not, on their own, share presence between their
+        # members. This method is entirely per-room presence fan-out, so skip
+        # excluded rooms wholesale.
+        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.
@@ -1929,6 +1956,9 @@ class PresenceEventSource(EventSource[int, UserPresenceState]):
         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,
@@ -2043,9 +2073,31 @@ class PresenceEventSource(EventSource[int, UserPresenceState]):
                         **{SERVER_NAME_LABEL: self.server_name},
                     ).inc()
 
-                    sharing_users = await self.store.do_users_share_a_room(
-                        user_id, updated_users
-                    )
+                    # An updated user is interesting if they share a
+                    # (non-excluded) room with the syncing user. We check by
+                    # intersecting the cached per-user room sets rather than via
+                    # `do_users_share_a_room`: its per-pair cache has a
+                    # quadratic working set and is cleared wholesale on every
+                    # membership change, so on busy servers every check missed
+                    # into SQL.
+                    #
+                    # For every presence update we need to run this code for
+                    # every user that is currently syncing. The
+                    # `get_rooms_for_user` will therefore be computed only once
+                    # for each updated user regardless of the number of syncing
+                    # users.
+                    #
+                    # The syncing user's rooms will also be cached as its needed
+                    # during sync processing anyway.
+                    my_rooms = await self.store.get_rooms_for_user(user_id)
+                    if self._rooms_to_exclude_from_presence:
+                        my_rooms = my_rooms - self._rooms_to_exclude_from_presence
+                    rooms_by_user = await self.store.get_rooms_for_users(updated_users)
+                    sharing_users = {
+                        updated_user
+                        for updated_user, rooms in rooms_by_user.items()
+                        if not my_rooms.isdisjoint(rooms)
+                    }
 
                     interested_and_updated_users = (
                         sharing_users.union(additional_users_interested_in)
@@ -2060,7 +2112,9 @@ class PresenceEventSource(EventSource[int, UserPresenceState]):
                     ).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)
 
@@ -2073,7 +2127,9 @@ class PresenceEventSource(EventSource[int, UserPresenceState]):
                 # 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)
 
@@ -2473,7 +2529,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.
@@ -2482,6 +2541,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)`,
@@ -2492,6 +2553,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
@@ -2512,6 +2575,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.
@@ -2522,6 +2586,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.
@@ -2535,6 +2601,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

@@ -971,12 +971,22 @@ class RoomMemberWorkerStore(EventsWorkerStore, CacheInvalidationWorkerStore):
 
         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)
 

tests/handlers/test_presence.py

@@ -51,6 +51,8 @@ from synapse.handlers.presence import (
     FEDERATION_TIMEOUT,
     PresenceHandler,
     WorkerPresenceHandler,
+    get_interested_parties,
+    get_interested_remotes,
     handle_timeout,
     handle_update,
 )
@@ -2323,6 +2325,217 @@ class PresenceJoinTestCase(unittest.HomeserverTestCase):
         return event
 
 
+class PresenceExcludeRoomsTestCase(unittest.HomeserverTestCase):
+    """Tests that `exclude_rooms_from_presence` stops presence being routed
+    between users solely because they share an excluded room."""
+
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+    ]
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.hs = hs
+        self.store = hs.get_datastores().main
+        self.presence_router = hs.get_presence_router()
+        self.presence_handler = hs.get_presence_handler()
+
+        self.user1 = self.register_user("user1", "pass")
+        self.token1 = self.login("user1", "pass")
+        self.user2 = self.register_user("user2", "pass")
+        self.token2 = self.login("user2", "pass")
+
+    def test_excluded_rooms_not_routed(self) -> None:
+        # Two rooms that user1 is joined to.
+        excluded_room = self.helper.create_room_as(self.user1, tok=self.token1)
+        shared_room = self.helper.create_room_as(self.user1, tok=self.token1)
+
+        state = UserPresenceState.default(self.user1)
+
+        # Without any exclusions both rooms are interested in user1's presence.
+        room_ids_to_states, users_to_states = self.get_success(
+            get_interested_parties(self.store, self.presence_router, [state])
+        )
+        self.assertIn(excluded_room, room_ids_to_states)
+        self.assertIn(shared_room, room_ids_to_states)
+
+        # Excluding one room drops it as an interested party, but the other
+        # (non-excluded) room still routes presence...
+        room_ids_to_states, users_to_states = self.get_success(
+            get_interested_parties(
+                self.store,
+                self.presence_router,
+                [state],
+                frozenset({excluded_room}),
+            )
+        )
+        self.assertNotIn(excluded_room, room_ids_to_states)
+        self.assertIn(shared_room, room_ids_to_states)
+
+        # ...and the user always receives their own presence, even when all of
+        # their rooms are excluded.
+        room_ids_to_states, users_to_states = self.get_success(
+            get_interested_parties(
+                self.store,
+                self.presence_router,
+                [state],
+                frozenset({excluded_room, shared_room}),
+            )
+        )
+        self.assertNotIn(excluded_room, room_ids_to_states)
+        self.assertNotIn(shared_room, room_ids_to_states)
+        self.assertIn(self.user1, users_to_states)
+
+    @override_config({"exclude_rooms_from_presence": ["!excluded:test"]})
+    def test_config_populates_handler(self) -> None:
+        """The config option should be plumbed through to the presence handler
+        and the presence event source as a frozenset."""
+        self.assertEqual(
+            self.presence_handler._rooms_to_exclude_from_presence,
+            frozenset({"!excluded:test"}),
+        )
+
+        event_source = self.hs.get_event_sources().sources.presence
+        self.assertEqual(
+            event_source._rooms_to_exclude_from_presence,
+            frozenset({"!excluded:test"}),
+        )
+
+    def test_is_visible_respects_excluded_rooms(self) -> None:
+        """`is_visible` (which drives the read side of /sync) should not
+        consider two users to share presence solely via an excluded room."""
+        user1 = UserID.from_string(self.user1)
+        user2 = UserID.from_string(self.user2)
+
+        # A single shared room: the two users can see each other's presence.
+        excluded_room = self.helper.create_room_as(self.user1, tok=self.token1)
+        self.helper.join(excluded_room, self.user2, tok=self.token2)
+
+        self.assertTrue(
+            self.get_success(self.presence_handler.is_visible(user2, user1))
+        )
+
+        # Excluding the only shared room hides presence between them.
+        self.presence_handler._rooms_to_exclude_from_presence = frozenset(
+            {excluded_room}
+        )
+        self.assertFalse(
+            self.get_success(self.presence_handler.is_visible(user2, user1))
+        )
+
+        # But a second, non-excluded shared room restores visibility.
+        shared_room = self.helper.create_room_as(self.user1, tok=self.token1)
+        self.helper.join(shared_room, self.user2, tok=self.token2)
+        self.assertTrue(
+            self.get_success(self.presence_handler.is_visible(user2, user1))
+        )
+
+    def test_get_interested_remotes_respects_excluded_rooms(self) -> None:
+        """The federation fan-out side (`get_interested_remotes`) must not route
+        presence to servers reached solely via an excluded room."""
+        excluded_room = self.helper.create_room_as(self.user1, tok=self.token1)
+        state = UserPresenceState.default(self.user1)
+
+        def hosts_for(excluded: frozenset) -> set:
+            result = self.get_success(
+                get_interested_remotes(
+                    self.store, self.presence_router, [state], excluded
+                )
+            )
+            hosts: set[str] = set()
+            for room_hosts, _ in result:
+                hosts.update(room_hosts)
+            return hosts
+
+        # The local server is a host in the room (all members are local here),
+        # so presence would be routed there...
+        self.assertIn("test", hosts_for(frozenset()))
+        # ...but excluding the only room removes it as a source of destinations.
+        self.assertNotIn("test", hosts_for(frozenset({excluded_room})))
+
+
+class PresenceGetNewEventsStreamTestCase(unittest.HomeserverTestCase):
+    """Tests the incremental (`from_key`) branch of
+    `PresenceEventSource.get_new_events`, which decides which updated users are
+    interesting to the syncing user by intersecting their cached room sets.
+    """
+
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+    ]
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        self.presence_handler = hs.get_presence_handler()
+        self.event_source = hs.get_event_sources().sources.presence
+
+        self.user1 = self.register_user("user1", "pass")
+        self.token1 = self.login("user1", "pass")
+        self.user2 = self.register_user("user2", "pass")
+        self.token2 = self.login("user2", "pass")
+        self.user3 = self.register_user("user3", "pass")
+        self.token3 = self.login("user3", "pass")
+
+    def _set_presence(self, user_id: str, state: str = "online") -> None:
+        self.get_success(
+            self.presence_handler.set_state(
+                UserID.from_string(user_id), "dev", {"presence": state}
+            )
+        )
+
+    def _updated_users_seen_by(self, user_id: str, from_key: int) -> set[str]:
+        states, _ = self.get_success(
+            self.event_source.get_new_events(
+                user=UserID.from_string(user_id), from_key=from_key
+            )
+        )
+        return {state.user_id for state in states}
+
+    def test_incremental_interest(self) -> None:
+        """A syncing user sees updates from users they share a room with (and
+        themselves), but not from strangers."""
+        shared_room = self.helper.create_room_as(self.user1, tok=self.token1)
+        self.helper.join(shared_room, self.user2, tok=self.token2)
+        # user3 is in an unrelated room.
+        self.helper.create_room_as(self.user3, tok=self.token3)
+
+        from_key = self.event_source.get_current_key()
+        self._set_presence(self.user1)
+        self._set_presence(self.user2)
+        self._set_presence(self.user3)
+
+        seen = self._updated_users_seen_by(self.user2, from_key)
+        self.assertIn(self.user1, seen)
+        self.assertIn(self.user2, seen)  # always sees own updates
+        self.assertNotIn(self.user3, seen)
+
+    def test_incremental_interest_excluded_room(self) -> None:
+        """Sharing only an excluded room does not make an updated user
+        interesting; sharing an additional normal room does."""
+        excluded_room = self.helper.create_room_as(self.user1, tok=self.token1)
+        self.helper.join(excluded_room, self.user2, tok=self.token2)
+
+        self.event_source._rooms_to_exclude_from_presence = frozenset({excluded_room})
+
+        from_key = self.event_source.get_current_key()
+        self._set_presence(self.user1, "online")
+        seen = self._updated_users_seen_by(self.user2, from_key)
+        self.assertNotIn(self.user1, seen)
+
+        # A second, non-excluded shared room restores interest. (Use a
+        # different presence state, as repeating the same one would not
+        # generate a new update.)
+        shared_room = self.helper.create_room_as(self.user1, tok=self.token1)
+        self.helper.join(shared_room, self.user2, tok=self.token2)
+
+        from_key = self.event_source.get_current_key()
+        self._set_presence(self.user1, "unavailable")
+        seen = self._updated_users_seen_by(self.user2, from_key)
+        self.assertIn(self.user1, seen)
+
+
 class WorkerPresenceThrottleTestCase(BaseMultiWorkerStreamTestCase):
     """Tests that sync workers suppress the per-sync-request presence updates
     that the presence writer would discard anyway, while relaying genuine