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:
         """This is called by the notifier in the background
@@ -224,14 +224,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":
                     events = await self._handle_typing(service, new_token)
                     if events:
                         self.scheduler.submit_ephemeral_events_for_as(service, events)

synapse/notifier.py

@@ -381,11 +381,12 @@ def _notify_app_services_ephemeral(
         users: Optional[Collection[Union[str, UserID]]] = None,
     ):
         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")