Skip application services with url: null when notifying them of events (#20273)

ESS deployments of MatrixRTC are going to run `lk-jwt-service` as an
application service so that it can manage delayed events. However,
setting up an application service is particularly inefficient, as
Synapse iterates through all the events of the rooms it covers since
their creation.

When an application service is configured with `url: null`, there is no
need to traverse all the events for it, so we can set up such
application services much faster without having to decide what to do for
other application services. Discussions about those are still open:
element-hq/synapse#11629 (see also matrix-org/synapse#1834).


### Pull Request Checklist

<!-- Please read
https://element-hq.github.io/synapse/latest/development/contributing_guide.html
before submitting your pull request -->

* [x] Pull request is based on the develop branch
* [x] Pull request includes a [changelog
file](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#changelog).
The entry should:
- Be a short description of your change which makes sense to users.
"Fixed a bug that prevented receiving messages from other servers."
instead of "Moved X method from `EventStore` to `EventWorkerStore`.".
  - Use markdown where necessary, mostly for `code blocks`.
  - End with either a period (.) or an exclamation mark (!).
  - Start with a capital letter.
- Feel free to credit yourself, by adding a sentence "Contributed by
@github_username." or "Contributed by [Your Name]." to the end of the
entry.
* [x] [Code
style](https://element-hq.github.io/synapse/latest/code_style.html) is
correct (run the
[linters](https://element-hq.github.io/synapse/latest/development/contributing_guide.html#run-the-linters))

changelog.d/20273.bugfix

@@ -0,0 +1 @@
+Stop queueing events and ephemeral data for application services registered with `url: null` (such as bridges' double-puppeting registrations), which wasted work on every matching event and, when no other application service was registered, made Synapse read through its whole event history.

synapse/handlers/appservice.py

@@ -105,7 +105,7 @@ class ApplicationServicesHandler:
         # clock components.
         current_id = max_token.stream
 
-        services = self.store.get_app_services()
+        services = self._get_services_to_notify()
         if not services or not self.notify_appservices:
             return
 
@@ -300,7 +300,7 @@ class ApplicationServicesHandler:
         #
         # Note that whether these events are actually relevant to these appservices
         # is decided later on.
-        services = self.store.get_app_services()
+        services = self._get_services_to_notify()
         services = [
             service
             for service in services
@@ -846,14 +846,17 @@ class ApplicationServicesHandler:
     async def _get_services_for_event(
         self, event: EventBase
     ) -> list[ApplicationService]:
-        """Retrieve a list of application services interested in this event.
+        """Retrieve the application services interested in this event that we can
+        send it to. Services without a `url` are skipped: see
+        `_get_services_to_notify`.
 
         Args:
             event: The event to check.
         Returns:
-            A list of services interested in this event based on the service regex.
+            A list of services with a `url` interested in this event based on the
+            service regex.
         """
-        services = self.store.get_app_services()
+        services = self._get_services_to_notify()
 
         # we can't use a list comprehension here. Since python 3, list
         # comprehensions use a generator internally. This means you can't yield
@@ -865,6 +868,20 @@ class ApplicationServicesHandler:
 
         return interested_list
 
+    def _get_services_to_notify(self) -> list[ApplicationService]:
+        """Retrieve the application services that we can send transactions to.
+
+        A service registered with a `url` of `null` has asked for no traffic
+        (https://spec.matrix.org/v1.19/application-service-api/#registration), so
+        there is no point gathering anything to send to it. It still owns its
+        namespaces: look-ups by namespace must keep using `get_app_services`.
+        """
+        return [
+            service
+            for service in self.store.get_app_services()
+            if service.url is not None
+        ]
+
     def _get_services_for_user(self, user_id: str) -> list[ApplicationService]:
         services = self.store.get_app_services()
         return [s for s in services if (s.is_interested_in_user(user_id))]

tests/handlers/test_appservice.py

@@ -133,6 +133,56 @@ class AppServiceHandlerTestCase(unittest.TestCase):
             interested_service, events=[event]
         )
 
+    def test_notify_interested_services_only_service_without_url(self) -> None:
+        """
+        Test that an application service without a `url` does not make the handler
+        read any events, as it cannot be sent any.
+        """
+        service = self._mkservice(is_interested_in_event=True)
+        service.url = None
+        self.mock_store.get_app_services.return_value = [service]
+        self.mock_store.get_all_new_event_ids_stream = AsyncMock()
+        self.mock_store.get_events_as_list = AsyncMock()
+
+        self.handler.notify_interested_services(RoomStreamToken(stream=1))
+
+        self.mock_store.get_appservice_last_pos.assert_not_called()
+        self.mock_store.get_all_new_event_ids_stream.assert_not_called()
+        self.mock_store.get_events_as_list.assert_not_called()
+        self.mock_scheduler.enqueue_for_appservice.assert_not_called()
+        self.mock_store.set_appservice_last_pos.assert_not_called()
+
+    def test_notify_interested_services_skips_service_without_url(self) -> None:
+        """
+        Test that events are only sent to the application services that have a `url`,
+        and that the interest of a service without a `url` is not even checked.
+        """
+        service_with_url = self._mkservice(is_interested_in_event=True)
+        service_without_url = self._mkservice(is_interested_in_event=True)
+        service_without_url.url = None
+        self.mock_store.get_app_services.return_value = [
+            service_without_url,
+            service_with_url,
+        ]
+        self.mock_store.get_user_by_id = AsyncMock(
+            return_value={"name": "@someone:anywhere"}
+        )
+
+        event = Mock(
+            sender="@someone:anywhere", type="m.room.message", room_id="!foo:bar"
+        )
+        self.mock_store.get_all_new_event_ids_stream = AsyncMock(
+            return_value=(1, {event.event_id: 0})
+        )
+        self.mock_store.get_events_as_list = AsyncMock(return_value=[event])
+
+        self.handler.notify_interested_services(RoomStreamToken(stream=1))
+
+        service_without_url.is_interested_in_event.assert_not_called()
+        self.mock_scheduler.enqueue_for_appservice.assert_called_once_with(
+            service_with_url, events=[event]
+        )
+
     def test_query_user_exists_unknown_user(self) -> None:
         user_id = "@someone:anywhere"
         services = [self._mkservice(is_interested_in_event=True)]
@@ -385,6 +435,31 @@ class AppServiceHandlerTestCase(unittest.TestCase):
             interested_service, ephemeral=[]
         )
 
+    def test_notify_interested_services_ephemeral_skips_service_without_url(
+        self,
+    ) -> None:
+        """
+        Test that no ephemeral events are gathered for an application service without
+        a `url`, and that its stream position is left alone.
+        """
+        service = self._mkservice(is_interested_in_event=True)
+        service.url = None
+        self.mock_store.get_app_services.return_value = [service]
+        self.mock_store.get_type_stream_id_for_appservice = AsyncMock(return_value=579)
+        self.event_source.sources.receipt.get_new_events_as = AsyncMock(
+            return_value=([Mock(event_id="event_1")], None)
+        )
+
+        self.handler.notify_interested_services_ephemeral(
+            StreamKeyType.RECEIPT,
+            MultiWriterStreamToken(stream=580),
+            ["@fakerecipient:example.com"],
+        )
+
+        self.event_source.sources.receipt.get_new_events_as.assert_not_called()
+        self.mock_scheduler.enqueue_for_appservice.assert_not_called()
+        self.mock_store.set_appservice_stream_type_pos.assert_not_called()
+
     def _mkservice(
         self, is_interested_in_event: bool, protocols: Iterable | None = None
     ) -> Mock:
@@ -1413,6 +1488,9 @@ class ApplicationServicesHandlerSendEventsTestCase(unittest.HomeserverTestCase):
             rate_limited=False,
             namespaces=namespaces,
             supports_ephemeral=True,
+            # Must be set for Synapse to try pushing data to the AS
+            hs_token="abcde",
+            url="some_url",
         )
 
         # Register the application service
@@ -1561,6 +1639,9 @@ class ApplicationServicesHandlerOtkCountsTestCase(unittest.HomeserverTestCase):
                 ]
             },
             msc3202_transaction_extensions=True,
+            # Must be set for Synapse to try pushing data to the AS
+            hs_token="abcde",
+            url="some_url",
         )
         self.hs.get_datastores().main.services_cache = [self._service]