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 <oliverw@matrix.org>

changelog.d/20019.feature

@@ -0,0 +1 @@
+Add Synapse Module API hook that notifies modules when events are delivered over federation (`register_federation_callbacks(...)`)..

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)

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.

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)}

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:

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.
 

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)

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:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+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"
+                )

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:
+# <https://www.gnu.org/licenses/agpl-3.0.html>.
+#
+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)