Stop the master relaying USER_SYNC for other workers

synapse/app/generic_worker.py

@@ -38,7 +38,11 @@
 from synapse.config.logger import setup_logging
 from synapse.federation import send_queue
 from synapse.federation.transport.server import TransportLayerServer
-from synapse.handlers.presence import PresenceHandler, get_interested_parties
+from synapse.handlers.presence import (
+    AbstractPresenceHandler,
+    PresenceHandler,
+    get_interested_parties,
+)
 from synapse.http.server import JsonResource
 from synapse.http.servlet import RestServlet, parse_json_object_from_request
 from synapse.http.site import SynapseSite
@@ -224,7 +228,7 @@ async def on_POST(self, request, device_id):
 UPDATE_SYNCING_USERS_MS = 10 * 1000
 
 
-class GenericWorkerPresence(object):
+class GenericWorkerPresence(AbstractPresenceHandler):
     def __init__(self, hs):
         self.hs = hs
         self.is_mine_id = hs.is_mine_id

synapse/handlers/presence.py

@@ -22,7 +22,7 @@
     - PresenceHandler._handle_timeouts
     - should_notify
 """
-
+import abc
 import logging
 from contextlib import contextmanager
 from typing import Dict, Iterable, List, Set
@@ -42,7 +42,7 @@
 from synapse.metrics import LaterGauge
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.storage.presence import UserPresenceState
-from synapse.types import UserID, get_domain_from_id
+from synapse.types import JsonDict, UserID, get_domain_from_id
 from synapse.util.async_helpers import Linearizer
 from synapse.util.caches.descriptors import cached
 from synapse.util.metrics import Measure
@@ -100,7 +100,69 @@
 assert LAST_ACTIVE_GRANULARITY < IDLE_TIMER
 
 
-class PresenceHandler(object):
+class AbstractPresenceHandler(abc.ABC):
+    """The base interface for things implementing PresenceHandler"""
+
+    @abc.abstractmethod
+    async def user_syncing(
+        self, user_id: str, affect_presence: bool = True
+    ) -> ContextManager[None]:
+        """Returns a context manager that should surround any stream requests
+        from the user.
+
+        This allows us to keep track of who is currently streaming and who isn't
+        without having to have timers outside of this module to avoid flickering
+        when users disconnect/reconnect.
+
+        Args:
+            user_id (str)
+            affect_presence (bool): If false this function will be a no-op.
+                Useful for streams that are not associated with an actual
+                client that is being used by a user.
+        """
+
+    @abc.abstractmethod
+    def get_currently_syncing_users(self) -> Set[str]:
+        """Get the set of user ids that are currently syncing on this HS.
+        Returns:
+            set(str): A set of user_id strings.
+        """
+
+    @abc.abstractmethod
+    async def current_state_for_users(
+        self, user_ids: Iterable[str]
+    ) -> Dict[str, UserPresenceState]:
+        """Get the current presence state for multiple users.
+
+        Returns:
+            dict: `user_id` -> `UserPresenceState`
+        """
+
+    @abc.abstractmethod
+    async def get_state(self, target_user: UserID) -> UserPresenceState:
+        ...
+
+    @abc.abstractmethod
+    async def get_states(
+        self, target_user_ids: Iterable[str]
+    ) -> List[UserPresenceState]:
+        """Get the presence state for users.
+
+        Args:
+            target_user_ids (list)
+
+        Returns:
+            list
+        """
+
+    @abc.abstractmethod
+    async def set_state(
+        self, target_user: UserID, state: JsonDict, ignore_status_msg: bool = False
+    ) -> None:
+        """Set the presence state of the user. """
+
+
+class PresenceHandler(AbstractPresenceHandler):
     def __init__(self, hs: "synapse.server.HomeServer"):
         self.hs = hs
         self.is_mine_id = hs.is_mine_id

synapse/server.pyi

@@ -97,7 +97,9 @@ class HomeServer(object):
         pass
     def get_notifier(self) -> synapse.notifier.Notifier:
         pass
-    def get_presence_handler(self) -> synapse.handlers.presence.PresenceHandler:
+    def get_presence_handler(
+        self,
+    ) -> synapse.handlers.presence.AbstractPresenceHandler:
         pass
     def get_clock(self) -> synapse.util.Clock:
         pass