From 36664cac4fdec38993c68078cd9a82452b56dca7 Mon Sep 17 00:00:00 2001 From: Olivier 'reivilibre Date: Thu, 30 Jul 2026 12:32:27 +0100 Subject: [PATCH] Add Module API hooks that notify modules when events are delivered over federation. (#20019) Closes: https://github.com/element-hq/synapse/issues/19904 --------- Signed-off-by: Olivier 'reivilibre --- changelog.d/20019.feature | 1 + docs/SUMMARY.md | 1 + docs/modules/federation_callbacks.md | 39 ++ synapse/federation/federation_server.py | 85 +++- .../federation/sender/transaction_manager.py | 43 +- synapse/module_api/__init__.py | 17 + synapse/module_api/callbacks/__init__.py | 3 + synapse/module_api/callbacks/federation.py | 207 ++++++++++ tests/module_api/test_federation_callbacks.py | 378 ++++++++++++++++++ 9 files changed, 754 insertions(+), 20 deletions(-) create mode 100644 changelog.d/20019.feature create mode 100644 docs/modules/federation_callbacks.md create mode 100644 synapse/module_api/callbacks/federation.py create mode 100644 tests/module_api/test_federation_callbacks.py diff --git a/changelog.d/20019.feature b/changelog.d/20019.feature new file mode 100644 index 0000000000..2744fb4a8d --- /dev/null +++ b/changelog.d/20019.feature @@ -0,0 +1 @@ +Add Synapse Module API hook that notifies modules when events are delivered over federation (`register_federation_callbacks(...)`).. diff --git a/docs/SUMMARY.md b/docs/SUMMARY.md index 980f51d078..709e1a01e2 100644 --- a/docs/SUMMARY.md +++ b/docs/SUMMARY.md @@ -50,6 +50,7 @@ - [Background update controller callbacks](modules/background_update_controller_callbacks.md) - [Account data callbacks](modules/account_data_callbacks.md) - [Add extra fields to client events unsigned section callbacks](modules/add_extra_fields_to_client_events_unsigned.md) + - [Federation callbacks](modules/federation_callbacks.md) - [Media repository callbacks](modules/media_repository_callbacks.md) - [Ratelimit callbacks](modules/ratelimit_callbacks.md) - [Porting a legacy module to the new interface](modules/porting_legacy_module.md) diff --git a/docs/modules/federation_callbacks.md b/docs/modules/federation_callbacks.md new file mode 100644 index 0000000000..09154e5640 --- /dev/null +++ b/docs/modules/federation_callbacks.md @@ -0,0 +1,39 @@ +# Federation callbacks + +Federation callbacks can be registered using the module API's `register_federation_callbacks` method. + +## Callbacks + +The available federation callbacks are: + +### `on_event_delivered_over_federation` + +_First introduced in Synapse v1.158.0_ + +```python +async def on_event_delivered_over_federation( + event: FederationEventDeliveryEvent, +) -> None: +``` + +Called when an event has been delivered over federation. +See `FederationEventDeliveryEvent` for detailed information available on the event. + +Note that depending on the specific method, delivery may not have +actually been acknowledged by the other homeserver. +See `FederatedEventDeliveryMethod` for details on which cases imply acknowledgment. + +Modules should anticipate more methods being added to the `FederatedEventDeliveryMethod` enum +over time (it is non-exhaustive). + +Only methods that deliver full, signed PDUs are included in this mechanism. +Some notable examples of excluded endpoints: +- `/send_knock` is excluded as it only returns unsigned 'stripped state'. +- `/timestamp_to_event` is excluded as it only returns event IDs, not events themselves. + +If multiple modules implement this callback, Synapse runs them all in order. +Exceptions are logged and otherwise ignored. + +Performance note: +- Registering this hook causes a performance (caching) optimisation on the + Federation `/state` endpoint to be bypassed. diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 83566050f6..2379b2ed2c 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -81,6 +81,7 @@ from synapse.logging.opentracing import ( ) from synapse.metrics import SERVER_NAME_LABEL from synapse.metrics.background_process_metrics import wrap_as_background_process +from synapse.module_api.callbacks.federation import FederatedEventDeliveryMethod from synapse.replication.http.federation import ( ReplicationFederationSendEduRestServlet, ) @@ -142,6 +143,7 @@ class FederationServer(FederationBase): self.server_name = hs.hostname self.handler = hs.get_federation_handler() self._spam_checker_module_callbacks = hs.get_module_api_callbacks().spam_checker + self._federation_callbacks = hs.get_module_api_callbacks().federation self._federation_event_handler = hs.get_federation_event_handler() self.state = hs.get_state_handler() self._event_auth_handler = hs.get_event_auth_handler() @@ -244,6 +246,10 @@ class FederationServer(FederationBase): res = self._transaction_dict_from_pdus(pdus) + await self._federation_callbacks.notify_on_event_delivered_over_federation( + origin, pdus, FederatedEventDeliveryMethod.BACKFILL + ) + return 200, res async def on_timestamp_to_event_request( @@ -655,14 +661,27 @@ class FederationServer(FederationBase): # - but that's non-trivial to get right, and anyway somewhat defeats # the point of the linearizer. async with self._server_linearizer.queue((origin, room_id)): - resp = await self._state_resp_cache.wrap( - (room_id, event_id), - self._on_context_state_request_compute, - room_id, - event_id, - ) - - return 200, resp + if not self._federation_callbacks.interested_in_events_delivered_over_federation(): + # In the usual case where no module is interested in tracking event deliveries, + # use the response cache. + resp = await self._state_resp_cache.wrap( + (room_id, event_id), + self._on_context_state_request_compute, + room_id, + event_id, + ) + return 200, resp + else: + # When a module is interested in tracking event deliveries, + # we can't use the response cache that returns pre-serialised + # events, as we wouldn't have the raw events to track. + resp, events = await self._on_context_state_request_compute_with_events( + room_id, event_id + ) + await self._federation_callbacks.notify_on_event_delivered_over_federation( + origin, events, FederatedEventDeliveryMethod.STATE + ) + return 200, resp @trace @tag_args @@ -697,6 +716,28 @@ class FederationServer(FederationBase): async def _on_context_state_request_compute( self, room_id: str, event_id: str ) -> dict[str, list]: + """ + Respond to a `/state` request, returning just the response. + + This separation exists because we don't want to hold on to the underlying + events in the response cache, just the serialised JSON. + """ + resp, _ = await self._on_context_state_request_compute_with_events( + room_id, event_id + ) + return resp + + async def _on_context_state_request_compute_with_events( + self, room_id: str, event_id: str + ) -> tuple[dict[str, list], list[EventBase]]: + """ + Respond to a `/state` request. + + Returns: + Tuple of: + 1. the `/state` response + 2. list of the events used to build that response + """ pdus: Collection[EventBase] event_ids = await self.handler.get_state_ids_for_pdu(room_id, event_id) pdus = await self.store.get_events_as_list(event_ids) @@ -705,10 +746,13 @@ class FederationServer(FederationBase): room_id, [pdu.event_id for pdu in pdus] ) - return { - "pdus": serialize_and_filter_pdus(pdus), - "auth_chain": serialize_and_filter_pdus(auth_chain), - } + return ( + { + "pdus": serialize_and_filter_pdus(pdus), + "auth_chain": serialize_and_filter_pdus(auth_chain), + }, + [*pdus, *auth_chain], + ) async def on_pdu_request( self, origin: str, event_id: str @@ -716,6 +760,9 @@ class FederationServer(FederationBase): pdu = await self.handler.get_persisted_pdu(origin, event_id) if pdu: + await self._federation_callbacks.notify_on_event_delivered_over_federation( + origin, [pdu], FederatedEventDeliveryMethod.EVENT + ) return 200, self._transaction_dict_from_pdus([pdu]) else: return 404, "" @@ -893,6 +940,12 @@ class FederationServer(FederationBase): if servers_in_room is not None: resp["servers_in_room"] = list(servers_in_room) + await self._federation_callbacks.notify_on_event_delivered_over_federation( + origin, + [event, *state_events, *auth_chain_events], + FederatedEventDeliveryMethod.SEND_JOIN, + ) + return resp async def on_make_leave_request( @@ -1131,6 +1184,10 @@ class FederationServer(FederationBase): time_now = self._clock.time_msec() auth_pdus = await self.handler.on_event_auth(event_id, room_id) res = {"auth_chain": serialize_and_filter_pdus(auth_pdus, time_now)} + + await self._federation_callbacks.notify_on_event_delivered_over_federation( + origin, auth_pdus, FederatedEventDeliveryMethod.EVENT_AUTH + ) return 200, res async def on_query_client_keys( @@ -1213,6 +1270,10 @@ class FederationServer(FederationBase): else: logger.debug("Returning %d events", len(missing_events)) + await self._federation_callbacks.notify_on_event_delivered_over_federation( + origin, missing_events, FederatedEventDeliveryMethod.GET_MISSING_EVENTS + ) + time_now = self._clock.time_msec() return {"events": serialize_and_filter_pdus(missing_events, time_now)} diff --git a/synapse/federation/sender/transaction_manager.py b/synapse/federation/sender/transaction_manager.py index 99aa05ebd6..9a7fb8a3b1 100644 --- a/synapse/federation/sender/transaction_manager.py +++ b/synapse/federation/sender/transaction_manager.py @@ -18,7 +18,7 @@ # # import logging -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Mapping from prometheus_client import Gauge @@ -35,6 +35,7 @@ from synapse.logging.opentracing import ( whitelisted_homeserver, ) from synapse.metrics import SERVER_NAME_LABEL +from synapse.module_api.callbacks.federation import FederatedEventDeliveryMethod from synapse.types import JsonDict from synapse.util.json import json_decoder from synapse.util.metrics import measure_func @@ -64,6 +65,7 @@ class TransactionManager: self._store = hs.get_datastores().main self._transaction_actions = TransactionActions(self._store) self._transport_layer = hs.get_federation_transport_client() + self._federation_callbacks = hs.get_module_api_callbacks().federation self._federation_metrics_domains = ( hs.config.federation.federation_metrics_domains @@ -192,14 +194,39 @@ class TransactionManager: logger.info("TX [%s] {%s} got 200 response", destination, txn_id) - for e_id, r in response.get("pdus", {}).items(): - if "error" in r: - logger.warning( - "TX [%s] {%s} Remote returned error for %s: %s", + pdu_responses = response.get("pdus", {}) + if not isinstance(pdu_responses, Mapping): + logger.warning( + "TX [%s] {%s} Remote returned invalid type for `pdus`", + destination, + txn_id, + ) + else: + for event_id, pdu_response in pdu_responses.items(): + if not isinstance(pdu_response, Mapping) or "error" in pdu_response: + logger.warning( + "TX [%s] {%s} Remote returned error for %s: %s", + destination, + txn_id, + event_id, + pdu_response, + ) + + # If modules have requested to be notified about delivered events, + # build and send that notification. + if self._federation_callbacks.interested_in_events_delivered_over_federation(): + # A PDU is considered acknowledged when the remote echoes the event_id back to + # us, without an error in the PDU response dict. + acknowledged_pdu_ids = { + event_id + for event_id, pdu_response in response.get("pdus", {}).items() + if isinstance(pdu_response, Mapping) + and "error" not in pdu_response + } + await self._federation_callbacks.notify_on_event_delivered_over_federation( destination, - txn_id, - e_id, - r, + [p for p in pdus if p.event_id in acknowledged_pdu_ids], + FederatedEventDeliveryMethod.SEND, ) if pdus and destination in self._federation_metrics_domains: diff --git a/synapse/module_api/__init__.py b/synapse/module_api/__init__.py index 1131253028..48963b8d83 100644 --- a/synapse/module_api/__init__.py +++ b/synapse/module_api/__init__.py @@ -90,6 +90,9 @@ from synapse.module_api.callbacks.account_validity_callbacks import ( ON_USER_LOGIN_CALLBACK, ON_USER_REGISTRATION_CALLBACK, ) +from synapse.module_api.callbacks.federation import ( + ON_EVENT_DELIVERED_OVER_FEDERATION_CALLBACK, +) from synapse.module_api.callbacks.media_repository_callbacks import ( GET_MEDIA_CONFIG_FOR_USER_CALLBACK, GET_MEDIA_UPLOAD_LIMITS_FOR_USER_CALLBACK, @@ -631,6 +634,20 @@ class ModuleApi: add_field_to_unsigned_callback ) + def register_federation_callbacks( + self, + *, + on_event_delivered_over_federation: ON_EVENT_DELIVERED_OVER_FEDERATION_CALLBACK + | None = None, + ) -> None: + """Registers callbacks for federation. + + Added in Synapse v1.158.0.""" + if on_event_delivered_over_federation is not None: + self._callbacks.federation.register_callbacks( + on_event_delivered_over_federation=on_event_delivered_over_federation + ) + ######################################################################### # The following methods can be called by the module at any point in time. diff --git a/synapse/module_api/callbacks/__init__.py b/synapse/module_api/callbacks/__init__.py index 16ef7a4b47..6ce429dd1c 100644 --- a/synapse/module_api/callbacks/__init__.py +++ b/synapse/module_api/callbacks/__init__.py @@ -21,6 +21,8 @@ from typing import TYPE_CHECKING +from synapse.module_api.callbacks.federation import FederationModuleApiCallbacks + if TYPE_CHECKING: from synapse.server import HomeServer @@ -44,6 +46,7 @@ from synapse.module_api.callbacks.third_party_event_rules_callbacks import ( class ModuleApiCallbacks: def __init__(self, hs: "HomeServer") -> None: self.account_validity = AccountValidityModuleApiCallbacks() + self.federation = FederationModuleApiCallbacks() self.media_repository = MediaRepositoryModuleApiCallbacks(hs) self.ratelimit = RatelimitModuleApiCallbacks(hs) self.spam_checker = SpamCheckerModuleApiCallbacks(hs) diff --git a/synapse/module_api/callbacks/federation.py b/synapse/module_api/callbacks/federation.py new file mode 100644 index 0000000000..ce879b764b --- /dev/null +++ b/synapse/module_api/callbacks/federation.py @@ -0,0 +1,207 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 Element Creations Ltd +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# See the GNU Affero General Public License for more details: +# . +# +import logging +from enum import Enum +from typing import Awaitable, Callable, Collection + +import attr + +from synapse.events import EventBase + +logger = logging.getLogger(__name__) + + +class FederatedEventDeliveryMethod(str, Enum): + """ + Method by which an event was 'delivered' to another server. + + Note that depending on the specific method, delivery may not have + actually been acknowledged by the other homeserver. + + Modules should anticipate more methods being added to this enum + over time (it is non-exhaustive). + + Only methods that deliver full, signed PDUs are included in this mechanism. + Some notable examples of excluded endpoints: + - `/send_knock` is excluded as it only returns unsigned 'stripped state'. + - `/timestamp_to_event` is excluded as it only returns event IDs, not events themselves. + """ + + SEND = "/send" + """ + The events were pushed over [`/send`](https://spec.matrix.org/v1.19/server-server-api/#put_matrixfederationv1sendtxnid). + + When a callback is triggered with this method, the events have been acknowledged + without error. + """ + + BACKFILL = "/backfill" + """ + The events were pulled over [`/backfill`](https://spec.matrix.org/v1.19/server-server-api/#get_matrixfederationv1backfillroomid). + + When a callback is triggered with this method, the events have _not_ been + acknowledged by the remote. + Actual delivery depends on network conditions and other factors influencing + the successful processing of the response at the remote homeserver. + """ + + GET_MISSING_EVENTS = "/get_missing_events" + """ + The events were pulled over [`/get_missing_events`](https://spec.matrix.org/v1.19/server-server-api/#post_matrixfederationv1get_missing_eventsroomid). + + When a callback is triggered with this method, the events have _not_ been + acknowledged by the remote. + Actual delivery depends on network conditions and other factors influencing + the successful processing of the response at the remote homeserver. + """ + + EVENT = "/event" + """ + The event was pulled over [`/event`](https://spec.matrix.org/v1.19/server-server-api/#get_matrixfederationv1eventeventid). + + When a callback is triggered with this method, the events have _not_ been + acknowledged by the remote. + Actual delivery depends on network conditions and other factors influencing + the successful processing of the response at the remote homeserver. + """ + + EVENT_AUTH = "/event_auth" + """ + The events were pulled over [`/event_auth`](https://spec.matrix.org/v1.19/server-server-api/#get_matrixfederationv1event_authroomideventid). + + When a callback is triggered with this method, the events have _not_ been + acknowledged by the remote. + Actual delivery depends on network conditions and other factors influencing + the successful processing of the response at the remote homeserver. + """ + + STATE = "/state" + """ + The events were pulled over [`/state`](https://spec.matrix.org/v1.19/server-server-api/#get_matrixfederationv1stateroomid). + + When a callback is triggered with this method, the events have _not_ been + acknowledged by the remote. + Actual delivery depends on network conditions and other factors influencing + the successful processing of the response at the remote homeserver. + """ + + SEND_JOIN = "/send_join" + """ + The events were pulled over [`/send_join`](https://spec.matrix.org/v1.19/server-server-api/#put_matrixfederationv2send_joinroomideventid). + + When a callback is triggered with this method, the events have _not_ been + acknowledged by the remote. + Actual delivery depends on network conditions and other factors influencing + the successful processing of the response at the remote homeserver. + """ + + +@attr.s(frozen=True, slots=True, auto_attribs=True) +class FederationEventDeliveryEvent: + """ + Represents the delivery of some events. + + Note that depending on `method`, + delivery may not be acknowledged. + """ + + server_name: str + """ + The server name of the destination the events were delivered to. + """ + + events: Collection[EventBase] + """ + The events that were delivered. + + Modules should not rely on this being the exhaustive list of all events that + were delivered in a single request; + delivery hooks may be triggered in multiple batches. + """ + + method: FederatedEventDeliveryMethod + """ + How the events were delivered to the server. + """ + + +ON_EVENT_DELIVERED_OVER_FEDERATION_CALLBACK = Callable[ + [FederationEventDeliveryEvent], Awaitable[None] +] + + +class FederationModuleApiCallbacks: + """ + Module API callbacks for generic federation events. + """ + + def __init__(self) -> None: + self._on_event_delivered_over_federation_callbacks: list[ + ON_EVENT_DELIVERED_OVER_FEDERATION_CALLBACK + ] = [] + + def interested_in_events_delivered_over_federation(self) -> bool: + """ + Whether any `on_event_delivered_over_federation` callbacks are registered. + """ + return len(self._on_event_delivered_over_federation_callbacks) > 0 + + def register_callbacks( + self, + on_event_delivered_over_federation: ON_EVENT_DELIVERED_OVER_FEDERATION_CALLBACK + | None = None, + ) -> None: + """ + Register callbacks from module for each hook. + + on_event_delivered_over_federation: + Callback fired when an event is delivered over federation. + See `FederationEventDeliveryEvent` for details. + + Performance note: + Registering this hook causes a performance (caching) optimisation on the + Federation `/state` endpoint to be bypassed. + """ + if on_event_delivered_over_federation is not None: + self._on_event_delivered_over_federation_callbacks.append( + on_event_delivered_over_federation + ) + + async def notify_on_event_delivered_over_federation( + self, + server_name: str, + events: Collection[EventBase], + method: FederatedEventDeliveryMethod, + ) -> None: + """Fire the registered callbacks to notify modules that some events were + delivered to another homeserver over federation. + + Does nothing if no callbacks are registered or if there are no events to + report. A callback that raises is logged and does not interrupt the others. + """ + if not events or not self._on_event_delivered_over_federation_callbacks: + return + + delivery = FederationEventDeliveryEvent( + server_name=server_name, + events=events, + method=method, + ) + for callback in self._on_event_delivered_over_federation_callbacks: + try: + await callback(delivery) + except Exception: + logger.exception( + "Error running on_event_delivered_over_federation callback" + ) diff --git a/tests/module_api/test_federation_callbacks.py b/tests/module_api/test_federation_callbacks.py new file mode 100644 index 0000000000..b8a34a0866 --- /dev/null +++ b/tests/module_api/test_federation_callbacks.py @@ -0,0 +1,378 @@ +# +# This file is licensed under the Affero General Public License (AGPL) version 3. +# +# Copyright (C) 2026 Element Creations Ltd +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. +# +# See the GNU Affero General Public License for more details: +# . +# +from http import HTTPStatus +from typing import Callable +from unittest.mock import AsyncMock, Mock + +from twisted.internet.testing import MemoryReactor + +from synapse.api.constants import EventTypes +from synapse.api.room_versions import KNOWN_ROOM_VERSIONS +from synapse.config.server import DEFAULT_ROOM_VERSION +from synapse.federation.federation_base import event_from_pdu_json +from synapse.federation.units import Transaction +from synapse.module_api.callbacks.federation import ( + FederatedEventDeliveryMethod, + FederationEventDeliveryEvent, +) +from synapse.rest import admin +from synapse.rest.client import login, room +from synapse.server import HomeServer +from synapse.types import JsonDict +from synapse.util.clock import Clock + +from tests import unittest + + +class FederationDeliveryCallbackTests(unittest.FederatingHomeserverTestCase): + """ + Tests for `on_event_delivered_over_federation` module callbacks. + """ + + servlets = [ + admin.register_servlets, + room.register_servlets, + login.register_servlets, + ] + + def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: + # Mock out the calls over federation. + self.fed_transport_client = Mock(spec=["send_transaction"]) + self.fed_transport_client.send_transaction = AsyncMock(return_value={}) + + hs = self.setup_test_homeserver( + federation_transport_client=self.fed_transport_client, + ) + + return hs + + def default_config(self) -> JsonDict: + # By default, federation sending is disabled in tests. + # Re-enable it for the main process. + config = super().default_config() + config["federation_sender_instances"] = None + return config + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + super().prepare(reactor, clock, hs) + + # Record every delivery the module callback is told about. + self._deliveries: list[FederationEventDeliveryEvent] = [] + + async def record(delivery: FederationEventDeliveryEvent) -> None: + self._deliveries.append(delivery) + + hs.get_module_api().register_federation_callbacks( + on_event_delivered_over_federation=record + ) + + # Create a public room with the remote server joined + self.creator = self.register_user("creator", "pass") + self.creator_tok = self.login("creator", "pass") + self.room_id = self.helper.create_room_as( + self.creator, tok=self.creator_tok, is_public=True + ) + self.remote_user = f"@remote:{self.OTHER_SERVER_NAME}" + self.inject_room_member(self.room_id, self.remote_user, "join") + + def _assert_only_delivery( + self, + method: FederatedEventDeliveryMethod, + ) -> FederationEventDeliveryEvent: + """ + Assert that exactly one delivery, with the given `method`, is currently recorded + (since the tracker was last cleared) and return it. + + This clears the tracker. + """ + self.assertEqual( + len(self._deliveries), + 1, + f"expected exactly one delivery; saw {self._deliveries!r}", + ) + + delivery = self._deliveries[0] + self._deliveries.clear() + + self.assertEqual(delivery.method, method, delivery) + + return delivery + + def test_backfill(self) -> None: + """ + Tests that the callback is triggered for incoming `/backfill` requests. + """ + ( + message_event_id1, + message_event_id2, + message_event_id3, + ) = self.helper.send_messages(self.room_id, 3, tok=self.creator_tok) + + # Call the endpoint twice to make sure that it doesn't forget to + # trigger the callback a second time, for example because it has + # a `ResponseCache` that bypasses the logic that triggers the + # callback. + for _ in range(2): + channel = self.make_signed_federation_request( + "GET", + f"/_matrix/federation/v1/backfill/{self.room_id}" + f"?v={message_event_id3}&limit=3", + ) + self.assertEqual(channel.code, HTTPStatus.OK, channel.result) + delivery = self._assert_only_delivery(FederatedEventDeliveryMethod.BACKFILL) + self.assertEqual(delivery.server_name, self.OTHER_SERVER_NAME) + self.assertEqual( + {e.event_id for e in delivery.events}, + {message_event_id1, message_event_id2, message_event_id3}, + ) + + def test_event(self) -> None: + """ + Tests that the callback is triggered for incoming `/event` requests. + """ + (message_event_id,) = self.helper.send_messages( + self.room_id, 1, tok=self.creator_tok + ) + + # Call the endpoint twice to make sure that it doesn't forget to + # trigger the callback a second time, for example because it has + # a `ResponseCache` that bypasses the logic that triggers the + # callback. + for _ in range(2): + channel = self.make_signed_federation_request( + "GET", f"/_matrix/federation/v1/event/{message_event_id}" + ) + self.assertEqual(channel.code, HTTPStatus.OK, channel.result) + delivery = self._assert_only_delivery(FederatedEventDeliveryMethod.EVENT) + self.assertIncludes( + {e.event_id for e in delivery.events}, + {message_event_id}, + exact=True, + ) + + def test_event_auth(self) -> None: + """ + Tests that the callback is triggered for incoming `/event_auth` requests. + """ + (message_event_id,) = self.helper.send_messages( + self.room_id, 1, tok=self.creator_tok + ) + + # Call the endpoint twice to make sure that it doesn't forget to + # trigger the callback a second time, for example because it has + # a `ResponseCache` that bypasses the logic that triggers the + # callback. + for _ in range(2): + channel = self.make_signed_federation_request( + "GET", + f"/_matrix/federation/v1/event_auth/{self.room_id}/{message_event_id}", + ) + self.assertEqual(channel.code, HTTPStatus.OK, channel.result) + delivery = self._assert_only_delivery( + FederatedEventDeliveryMethod.EVENT_AUTH + ) + + state_key_pairs_included = {(e.type, e.state_key) for e in delivery.events} + self.assertEqual( + state_key_pairs_included, + { + (EventTypes.Create, ""), + (EventTypes.PowerLevels, ""), + (EventTypes.Member, self.creator), + }, + ) + + def test_state(self) -> None: + """ + Tests that the callback is triggered for incoming `/state` requests. + """ + (message_event_id,) = self.helper.send_messages( + self.room_id, 1, tok=self.creator_tok + ) + + # Call the endpoint twice to make sure that it doesn't forget to + # trigger the callback a second time, for example because it has + # a `ResponseCache` that bypasses the logic that triggers the + # callback. + for _ in range(2): + channel = self.make_signed_federation_request( + "GET", + f"/_matrix/federation/v1/state/{self.room_id}?event_id={message_event_id}", + ) + self.assertEqual(channel.code, HTTPStatus.OK, channel.result) + delivery = self._assert_only_delivery(FederatedEventDeliveryMethod.STATE) + + # Check that we got notified about delivery for all the expected state events + # included in a `/state` response (including `pdus` and `auth_chain`) + state_key_pairs_included = {(e.type, e.state_key) for e in delivery.events} + self.assertEqual( + state_key_pairs_included, + { + (EventTypes.Create, ""), + (EventTypes.JoinRules, ""), + (EventTypes.PowerLevels, ""), + (EventTypes.RoomHistoryVisibility, ""), + (EventTypes.Member, self.creator), + (EventTypes.Member, self.remote_user), + }, + ) + + def test_get_missing_events(self) -> None: + """ + Tests that the callback is triggered for incoming `/get_missing_events` requests. + """ + ( + message_event_id1, + message_event_id2, + message_event_id3, + ) = self.helper.send_messages(self.room_id, 3, tok=self.creator_tok) + + # Call the endpoint twice to make sure that it doesn't forget to + # trigger the callback a second time, for example because it has + # a `ResponseCache` that bypasses the logic that triggers the + # callback. + for _ in range(2): + channel = self.make_signed_federation_request( + "POST", + f"/_matrix/federation/v1/get_missing_events/{self.room_id}", + { + "earliest_events": [message_event_id1], + "latest_events": [message_event_id3], + "limit": 10, + }, + ) + self.assertEqual(channel.code, HTTPStatus.OK, channel.result) + delivery = self._assert_only_delivery( + FederatedEventDeliveryMethod.GET_MISSING_EVENTS + ) + self.assertIncludes( + {e.event_id for e in delivery.events}, + {message_event_id2}, + exact=True, + ) + + def test_send_join(self) -> None: + """ + Tests that the callback is triggered for incoming `/send_join` requests, + including both the state events and the newly-created join event. + """ + joining_user = f"@joiner:{self.OTHER_SERVER_NAME}" + make_join = self.make_signed_federation_request( + "GET", + f"/_matrix/federation/v1/make_join/{self.room_id}/{joining_user}" + f"?ver={DEFAULT_ROOM_VERSION}", + ) + self.assertEqual(make_join.code, HTTPStatus.OK, make_join.json_body) + + join_event_dict = make_join.json_body["event"] + self.add_hashes_and_signatures_from_other_server( + join_event_dict, KNOWN_ROOM_VERSIONS[DEFAULT_ROOM_VERSION] + ) + channel = self.make_signed_federation_request( + "PUT", + f"/_matrix/federation/v2/send_join/{self.room_id}/x", + content=join_event_dict, + ) + self.assertEqual(channel.code, HTTPStatus.OK, channel.json_body) + + delivery = self._assert_only_delivery(FederatedEventDeliveryMethod.SEND_JOIN) + + # Check that we got notified about delivery for all the expected state events + # included in a `/send_join` response (the join itself, the room state + # and the auth chain events) + state_key_pairs_included = {(e.type, e.state_key) for e in delivery.events} + self.assertIncludes( + state_key_pairs_included, + { + (EventTypes.Create, ""), + (EventTypes.JoinRules, ""), + (EventTypes.PowerLevels, ""), + (EventTypes.RoomHistoryVisibility, ""), + (EventTypes.Member, self.creator), + (EventTypes.Member, self.remote_user), + (EventTypes.Member, joining_user), + }, + exact=True, + ) + + def test_send_outbound_transaction(self) -> None: + """ + Tests that the callback is triggered for outgoing `/send` transactions + when the remote acknowledges the PDU. + """ + + async def _acknowledge_pdus( + transaction: Transaction, + json_data_cb: Callable[[], JsonDict], + ) -> JsonDict: + """ + Acknowledge the PDUs. + """ + body = json_data_cb() + pdu_responses: JsonDict = {} + for pdu_json in body.get("pdus", []): + # We have to construct the event to calculate its event ID + event = event_from_pdu_json( + pdu_json, KNOWN_ROOM_VERSIONS[DEFAULT_ROOM_VERSION] + ) + # Empty dict means 'OK' + pdu_responses[event.event_id] = {} + return {"pdus": pdu_responses} + + self.fed_transport_client.send_transaction.side_effect = _acknowledge_pdus + + # After sending, the event propagates to the federation transmission queue + # and gets fired as a `/send` request + (message_event_id,) = self.helper.send_messages( + self.room_id, 1, tok=self.creator_tok + ) + + delivery = self._assert_only_delivery(FederatedEventDeliveryMethod.SEND) + self.assertEqual(delivery.server_name, self.OTHER_SERVER_NAME) + self.assertIncludes( + {e.event_id for e in delivery.events}, {message_event_id}, exact=True + ) + + def test_send_outbound_excludes_rejected_pdus(self) -> None: + """ + Tests that the event is NOT triggered for outgoing `/send` transactions + when the remote marks the PDU as failed. + """ + + async def _error_pdus( + transaction: Transaction, + json_data_cb: Callable[[], JsonDict], + ) -> JsonDict: + """ + Return an error for the PDUs. + """ + body = json_data_cb() + pdu_responses: JsonDict = {} + for pdu_json in body.get("pdus", []): + # We have to construct the event to calculate its event ID + event = event_from_pdu_json( + pdu_json, KNOWN_ROOM_VERSIONS[DEFAULT_ROOM_VERSION] + ) + pdu_responses[event.event_id] = {"error": "failed"} + return {"pdus": pdu_responses} + + self.fed_transport_client.send_transaction.side_effect = _error_pdus + + # After sending, the event propagates to the federation transmission queue + # and gets fired as a `/send` request + (_message_event_id,) = self.helper.send_messages( + self.room_id, 1, tok=self.creator_tok + ) + + self.assertIncludes(set(self._deliveries), set(), exact=True)