Fix providing a RoomStreamToken instance to _notify_app_services_ephemeral

changelog.d/11137.bugfix

@@ -0,0 +1 @@
+Partially address occasional bursts of old read receipt and presence data being sent to application services that have opted in to receiving them.

synapse/handlers/appservice.py

@@ -182,7 +182,7 @@ async def handle_room_events(events: Iterable[EventBase]) -> None:
     def notify_interested_services_ephemeral(
         self,
         stream_key: str,
-        new_token: Optional[int],
+        new_token: int,
         users: Optional[Collection[Union[str, UserID]]] = None,
     ) -> None:
         """
@@ -231,14 +231,13 @@ async def _notify_interested_services_ephemeral(
         self,
         services: List[ApplicationService],
         stream_key: str,
-        new_token: Optional[int],
+        new_token: int,
         users: Collection[Union[str, UserID]],
     ) -> None:
         logger.debug("Checking interested services for %s" % (stream_key))
         with Measure(self.clock, "notify_interested_services_ephemeral"):
             for service in services:
-                # Only handle typing if we have the latest token
-                if stream_key == "typing_key" and new_token is not None:
+                if stream_key == "typing_key":
                     # Note that we don't persist the token (via set_type_stream_id_for_appservice)
                     # for typing_key due to performance reasons and due to their highly
                     # ephemeral nature.

synapse/notifier.py

@@ -388,11 +388,12 @@ def _notify_app_services_ephemeral(
             users: The users that should be informed of the new event, if any.
         """
         try:
-            stream_token = None
-            if isinstance(new_token, int):
-                stream_token = new_token
+            # Convert new_token from a RoomStreamToken to an int if necessary
+            if isinstance(new_token, RoomStreamToken):
+                new_token = new_token.stream
+
             self.appservice_handler.notify_interested_services_ephemeral(
-                stream_key, stream_token, users or []
+                stream_key, new_token, users or []
             )
         except Exception:
             logger.exception("Error notifying application services of event")