Send transactions through Rust at rust_federation_sending: transactions

At the new `transactions` level, `TransactionManager` hands each
transaction to the Rust sender. Python keeps the tracing span, the
module callback and the `last_sent_pdu_time` gauge.

In tests, every transaction body Rust builds is checked against the one
Python builds (`SYNAPSE_TEST_RUST_FED_STRICT`). The federation sender
and catch-up test cases also run at the new level, with Rust building
the transactions that their mocked `send_transaction` receives.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01R5b7Ruy4cgt27Mjq6Ptxbb

changelog.d/rust-fed-txn.misc

@@ -0,0 +1 @@
+Build and send federation transactions in Rust at `experimental_features.rust_federation_sending: transactions`.

synapse/config/experimental.py

@@ -50,6 +50,17 @@ class MSC3866Config:
     require_approval_for_new_accounts: bool = False
 
 
+RustFederationSendingLevel = Literal["off", "client", "transactions"]
+
+# The levels of `rust_federation_sending`, in order. Each implies the ones
+# before it.
+RUST_FEDERATION_SENDING_LEVELS: tuple[RustFederationSendingLevel, ...] = (
+    "off",
+    "client",
+    "transactions",
+)
+
+
 class ExperimentalConfig(Config):
     """Config section for enabling experimental features
 
@@ -364,12 +375,23 @@ class ExperimentalConfig(Config):
         #    through the Rust request layer (signing, retries, the backoff and
         #    the connection pool). Media downloads, and an instance whose
         #    federation traffic goes through a proxy worker, stay Python.
+        #  - "transactions": the `TransactionManager` builds and sends `/send`
+        #    transactions in Rust, PDU JSON included.
         rust_federation_sending = experimental.get("rust_federation_sending", "off")
-        if rust_federation_sending not in ("off", "client"):
+        if rust_federation_sending not in RUST_FEDERATION_SENDING_LEVELS:
             raise ConfigError(
                 "experimental_features.rust_federation_sending must be one of "
-                "'off' or 'client'",
+                + ", ".join(repr(level) for level in RUST_FEDERATION_SENDING_LEVELS),
                 ("experimental", "rust_federation_sending"),
             )
-        self.rust_federation_sending: Literal["off", "client"] = rust_federation_sending
+        self.rust_federation_sending: RustFederationSendingLevel = (
+            rust_federation_sending
+        )
 
+    def rust_federation_sending_at_least(
+        self, level: RustFederationSendingLevel
+    ) -> bool:
+        """Whether `rust_federation_sending` is `level` or a later one."""
+        return RUST_FEDERATION_SENDING_LEVELS.index(
+            self.rust_federation_sending
+        ) >= RUST_FEDERATION_SENDING_LEVELS.index(level)

synapse/federation/sender/transaction_manager.py

@@ -17,6 +17,7 @@
 # [This file includes modifications made by New Vector Limited]
 #
 #
+import functools
 import logging
 from typing import TYPE_CHECKING, Mapping
 
@@ -27,6 +28,7 @@ from synapse.api.errors import HttpResponseException
 from synapse.events import EventBase
 from synapse.federation.persistence import TransactionActions
 from synapse.federation.units import Edu, Transaction, serialize_and_filter_pdus
+from synapse.logging import opentracing
 from synapse.logging.opentracing import (
     extract_text_map,
     set_tag,
@@ -36,6 +38,7 @@ from synapse.logging.opentracing import (
 )
 from synapse.metrics import SERVER_NAME_LABEL
 from synapse.module_api.callbacks.federation import FederatedEventDeliveryMethod
+from synapse.synapse_rust.federation import RustTransaction, RustTransactionManager
 from synapse.types import JsonDict
 from synapse.util.json import json_decoder
 from synapse.util.metrics import measure_func
@@ -57,8 +60,16 @@ class TransactionManager:
     """Helper class which handles building and sending transactions
 
     shared between PerDestinationQueue objects
+
+    At `rust_federation_sending: transactions` and above, Rust builds and sends
+    the transactions. This class keeps the tracing span and the module
+    callback.
     """
 
+    # For tests: check that each transaction Rust sends has the body Python
+    # would have sent.
+    _rust_strict = False
+
     def __init__(self, hs: "synapse.server.HomeServer"):
         self.server_name = hs.hostname  # nb must be called this for @measure_func
         self.clock = hs.get_clock()  # nb must be called this for @measure_func
@@ -76,6 +87,16 @@ class TransactionManager:
 
         self._is_shutdown = False
 
+        # The Rust transaction sender. It sends through the Rust request
+        # layer, so there is none on an instance whose federation traffic goes
+        # through a proxy worker.
+        self._rust: RustTransactionManager | None = None
+        self._is_mine_server_name = hs.is_mine_server_name
+        if hs.config.experimental.rust_federation_sending_at_least("transactions"):
+            rust_client = getattr(hs.get_federation_http_client(), "_rust_client", None)
+            if rust_client is not None:
+                self._rust = RustTransactionManager(rust_client)
+
     def shutdown(self) -> None:
         self._is_shutdown = True
         self._transport_layer.shutdown()
@@ -118,6 +139,10 @@ class TransactionManager:
         with start_active_span_follows_from("send_transaction", span_contexts):
             logger.debug("TX [%s] _attempt_new_transaction", destination)
 
+            if self._rust is not None:
+                await self._send_rust_transaction(self._rust, destination, pdus, edus)
+                return
+
             txn_id = str(self._next_txn_id)
 
             logger.debug(
@@ -235,3 +260,79 @@ class TransactionManager:
                     destination_server_name=destination,
                     **{SERVER_NAME_LABEL: self.server_name},
                 ).set(last_pdu.origin_server_ts / 1000)
+
+    async def _send_rust_transaction(
+        self,
+        rust: RustTransactionManager,
+        destination: str,
+        pdus: list[EventBase],
+        edus: list[Edu],
+    ) -> None:
+        """Have Rust build and send the transaction, and handle the response
+        as `send_new_transaction` does.
+        """
+        if self._is_mine_server_name(destination):
+            raise RuntimeError("Transport layer cannot send to itself!")
+
+        headers_dict: dict[bytes, list[bytes]] = {}
+        opentracing.inject_header_dict(headers_dict, destination)
+        extra_headers = [
+            (name.decode("ascii"), value.decode("ascii"))
+            for name, values in headers_dict.items()
+            for value in values
+        ]
+
+        want_acknowledged = (
+            self._federation_callbacks.interested_in_events_delivered_over_federation()
+        )
+        check = (
+            functools.partial(self._check_rust_transaction, pdus, edus)
+            if self._rust_strict
+            else None
+        )
+
+        try:
+            acknowledged = await rust.send_transaction(
+                destination,
+                pdus,
+                [edu.get_dict() for edu in edus],
+                extra_headers=extra_headers,
+                want_acknowledged=want_acknowledged,
+                check=check,
+            )
+        except HttpResponseException:
+            set_tag(tags.ERROR, True)
+            raise
+
+        if want_acknowledged and acknowledged is not None:
+            acknowledged_ids = set(acknowledged)
+            await self._federation_callbacks.notify_on_event_delivered_over_federation(
+                destination,
+                [p for p in pdus if p.event_id in acknowledged_ids],
+                FederatedEventDeliveryMethod.SEND,
+            )
+
+        if pdus and destination in self._federation_metrics_domains:
+            last_pdu = pdus[-1]
+            last_pdu_ts_metric.labels(
+                destination_server_name=destination,
+                **{SERVER_NAME_LABEL: self.server_name},
+            ).set(last_pdu.origin_server_ts / 1000)
+
+    def _check_rust_transaction(
+        self, pdus: list[EventBase], edus: list[Edu], transaction: RustTransaction
+    ) -> None:
+        """Raise if Rust built a different body from the one Python would."""
+        expected = Transaction(
+            origin_server_ts=transaction.origin_server_ts,
+            transaction_id=transaction.transaction_id,
+            origin=self.server_name,
+            destination=transaction.destination,
+            pdus=serialize_and_filter_pdus(pdus),
+            edus=[edu.get_dict() for edu in edus],
+        ).get_dict()
+        actual = transaction.get_dict()
+        if actual != expected:
+            raise AssertionError(
+                f"Rust transaction body differs from Python's: {actual!r} != {expected!r}"
+            )

synapse/http/matrixfederationclient.py

@@ -497,7 +497,7 @@ class MatrixFederationHttpClient:
         # client only talks to federation directly.
         self._rust_client: RustFederationClient | None = None
         if (
-            hs.config.experimental.rust_federation_sending != "off"
+            hs.config.experimental.rust_federation_sending_at_least("client")
             and hs.get_instance_name() in outbound_federation_restricted_to
         ):
             self._rust_client = RustFederationClient(

tests/federation/_rust_transactions.py

@@ -0,0 +1,99 @@
+#
+# 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>.
+#
+"""Run federation sender tests with Rust building the transactions.
+
+The tests mock `TransportLayerClient.send_transaction`, which the Rust sender
+does not call. `RustTransactionsMixin` puts a stand-in for the Rust sender in
+its place: Rust builds each transaction, the mock is called with it as it
+would be with a Python `Transaction`, and Rust reads the mock's response. The
+HTTP half of the Rust sender is tested in `tests.http.test_rust_transactions`.
+"""
+
+from typing import TYPE_CHECKING, Callable, Sequence
+
+from twisted.internet.testing import MemoryReactor
+
+from synapse.events import EventBase
+from synapse.federation.sender import FederationSender
+from synapse.server import HomeServer
+from synapse.synapse_rust.federation import RustTransaction, RustTransactionManager
+from synapse.types import JsonDict
+from synapse.util.clock import Clock
+
+if TYPE_CHECKING:
+    from unittest.mock import Mock
+
+    from tests.unittest import HomeserverTestCase
+
+    _Base = HomeserverTestCase
+else:
+    _Base = object
+
+
+class _MockedRustTransactions:
+    """Stands in for a `RustTransactionManager`, sending through the test's
+    mocked `send_transaction`."""
+
+    def __init__(self, rust: RustTransactionManager, transport: "Mock") -> None:
+        self._rust = rust
+        self._transport = transport
+
+    async def send_transaction(
+        self,
+        destination: str,
+        pdus: Sequence[EventBase],
+        edus: Sequence[JsonDict],
+        *,
+        extra_headers: Sequence[tuple[str, str]] = (),
+        want_acknowledged: bool = False,
+        check: Callable[[RustTransaction], None] | None = None,
+    ) -> list[str] | None:
+        transaction = self._rust.build_transaction(destination, pdus, edus)
+        if check is not None:
+            check(transaction)
+
+        response = await self._transport.send_transaction(
+            transaction, transaction.get_dict
+        )
+        if not isinstance(response, dict):
+            response = {}
+
+        acknowledged = self._rust.acknowledged_pdus(transaction, response)
+        return acknowledged if want_acknowledged else None
+
+
+class RustTransactionsMixin(_Base):
+    """Run a test case at `rust_federation_sending: transactions`. The test
+    case must mock the federation transport client as
+    `self.federation_transport_client`."""
+
+    federation_transport_client: "Mock"
+
+    def default_config(self) -> JsonDict:
+        config = super().default_config()
+        config.setdefault("experimental_features", {})["rust_federation_sending"] = (
+            "transactions"
+        )
+        return config
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        super().prepare(reactor, clock, hs)
+
+        sender = hs.get_federation_sender()
+        assert isinstance(sender, FederationSender)
+        manager = sender._transaction_manager
+        assert manager._rust is not None
+        manager._rust = _MockedRustTransactions(  # type: ignore[assignment]
+            manager._rust, self.federation_transport_client
+        )

tests/federation/test_federation_catch_up.py

@@ -22,6 +22,7 @@ from synapse.util.clock import Clock
 from synapse.util.duration import Duration
 from synapse.util.retryutils import NotRetryingDestination
 
+from tests.federation._rust_transactions import RustTransactionsMixin
 from tests.test_utils import event_injection
 from tests.unittest import FederatingHomeserverTestCase
 from tests.utils import USE_POSTGRES_FOR_TESTS
@@ -946,3 +947,14 @@ class FederationStickyEventCatchUpTestCase(_FederationCatchUpTestCaseBase):
         transaction = self.run_transaction(per_dest_queue)
         assert transaction is not None
         self.assertEqual([pdu.event_id for pdu in transaction.pdus], [sticky_id])
+
+
+class FederationCatchUpRustTestCases(RustTransactionsMixin, FederationCatchUpTestCases):
+    """As `FederationCatchUpTestCases`, with Rust building the transactions."""
+
+
+class FederationStickyEventCatchUpRustTestCase(
+    RustTransactionsMixin, FederationStickyEventCatchUpTestCase
+):
+    """As `FederationStickyEventCatchUpTestCase`, with Rust building the
+    transactions."""

tests/federation/test_federation_sender.py

@@ -39,6 +39,7 @@ from synapse.types import JsonDict, ReadReceipt
 from synapse.util.clock import Clock
 from synapse.util.duration import Duration
 
+from tests.federation._rust_transactions import RustTransactionsMixin
 from tests.unittest import HomeserverTestCase
 
 
@@ -966,3 +967,24 @@ def build_device_dict(user_id: str, device_id: str, sk: SigningKey) -> JsonDict:
             key_id(sk): encode_pubkey(sk),
         },
     }
+
+
+class FederationSenderReceiptsRustTestCases(
+    RustTransactionsMixin, FederationSenderReceiptsTestCases
+):
+    """As `FederationSenderReceiptsTestCases`, with Rust building the
+    transactions."""
+
+
+class FederationSenderPresenceRustTestCases(
+    RustTransactionsMixin, FederationSenderPresenceTestCases
+):
+    """As `FederationSenderPresenceTestCases`, with Rust building the
+    transactions."""
+
+
+class FederationSenderDevicesRustTestCases(
+    RustTransactionsMixin, FederationSenderDevicesTestCases
+):
+    """As `FederationSenderDevicesTestCases`, with Rust building the
+    transactions."""

tests/http/test_rust_transactions.py

@@ -0,0 +1,291 @@
+#
+# 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>.
+#
+"""Tests for `TransactionManager` at `rust_federation_sending: transactions`,
+where Rust builds and sends the transactions.
+
+As in `test_rust_federation_client`, the Rust client sends through a Python
+callable in place of the network. `TransactionManager._rust_strict` is set
+for every test, so each transaction body is also checked against the one
+Python builds.
+"""
+
+import json
+from typing import Any
+from unittest.mock import AsyncMock
+
+from twisted.internet.testing import MemoryReactor
+
+from synapse.api.errors import HttpResponseException, RequestSendFailed
+from synapse.events import EventBase
+from synapse.federation.sender.transaction_manager import (
+    TransactionManager,
+    last_pdu_ts_metric,
+)
+from synapse.federation.units import Edu
+from synapse.metrics import SERVER_NAME_LABEL
+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.synapse_rust.federation import (
+    RustFederationClient,
+    RustTransactionManager,
+    federation_gil_stats,
+)
+from synapse.util.clock import Clock
+
+from tests import unittest
+from tests.http.test_rust_federation_client import (
+    DEST,
+    FakeTransport,
+    json_response,
+)
+
+
+class RustTransactionsTestCase(unittest.HomeserverTestCase):
+    servlets = [
+        admin.register_servlets,
+        login.register_servlets,
+        room.register_servlets,
+    ]
+
+    def default_config(self) -> dict[str, Any]:
+        config = super().default_config()
+        config.setdefault("experimental_features", {})["rust_federation_sending"] = (
+            "transactions"
+        )
+        config["federation_metrics_domains"] = [DEST]
+        return config
+
+    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
+        cl = hs.get_federation_http_client()
+        self.transport = FakeTransport()
+        rust_client = RustFederationClient(
+            hs,
+            hs.version_string,
+            hs.version_string,
+            cl._write_retry_timings,
+            cl._record_request_metrics,
+            self.transport,
+        )
+
+        self.manager = TransactionManager(hs)
+        assert self.manager._rust is not None
+        self.manager._rust = RustTransactionManager(rust_client)
+
+        user = self.register_user("user", "pass")
+        tok = self.login("user", "pass")
+        room_id = self.helper.create_room_as(user, tok=tok)
+        self.helper.send(room_id, "hello", tok=tok)
+
+        store = hs.get_datastores().main
+        event_ids = self.get_success(store.get_latest_event_ids_in_room(room_id))
+        self.event = self.get_success(store.get_event(next(iter(event_ids))))
+
+    def _send(self, pdus: list[EventBase], edus: list[Edu] | None = None) -> None:
+        self.get_success(self.manager.send_new_transaction(DEST, pdus, edus or []))
+
+    def test_send(self) -> None:
+        """The transaction is PUT to `/send/<txn_id>` with the body Python
+        builds."""
+        self.transport.responses.append(json_response(200, {"pdus": {}}))
+        edu = Edu(
+            edu_type="m.typing",
+            content={"room_id": "!r:test", "user_id": "@u:test", "typing": True},
+            origin=self.hs.hostname,
+            destination=DEST,
+        )
+
+        self._send([self.event], [edu])
+
+        [request] = self.transport.requests
+        self.assertEqual(request.method, "PUT")
+        self.assertRegex(request.path, r"^/_matrix/federation/v1/send/\d+$")
+        self.assertIsNotNone(request.header(b"Authorization"))
+
+        body = json.loads(request.body)
+        self.assertEqual(body["origin"], self.hs.hostname)
+        self.assertEqual(body["edus"], [edu.get_dict()])
+        self.assertEqual(body["pdus"], [self.event.get_pdu_json()])
+
+    def test_txn_ids_increase(self) -> None:
+        self.transport.responses.extend([json_response(200, {})] * 2)
+        self._send([self.event])
+        self._send([self.event])
+
+        first, second = (
+            int(request.path.rsplit("/", 1)[1]) for request in self.transport.requests
+        )
+        self.assertEqual(second, first + 1)
+
+    def test_http_error(self) -> None:
+        """A 4xx is raised as Python raises it, and backs the destination
+        off."""
+        self.transport.responses.append(
+            json_response(403, {"errcode": "M_FORBIDDEN", "error": "no"})
+        )
+
+        f = self.get_failure(
+            self.manager.send_new_transaction(DEST, [self.event], []),
+            HttpResponseException,
+        )
+        self.assertEqual(f.value.code, 403)
+
+        # Let the dispatched write of the backoff run.
+        self.reactor.advance(0)
+        timings = self.get_success(
+            self.hs.get_datastores().main.get_destination_retry_timings(DEST)
+        )
+        self.assertIsNotNone(timings)
+
+    def test_invalid_response(self) -> None:
+        self.transport.responses.append(json_response(200, ["not", "a", "dict"]))
+        self.get_failure(
+            self.manager.send_new_transaction(DEST, [self.event], []),
+            RequestSendFailed,
+        )
+
+    def test_module_callback(self) -> None:
+        """Modules hear of the PDUs the remote accepted, and of no others."""
+        callback = AsyncMock(return_value=None)
+        self.hs.get_module_api_callbacks().federation.register_callbacks(
+            on_event_delivered_over_federation=callback
+        )
+        self.transport.responses.append(
+            json_response(
+                200,
+                {"pdus": {self.event.event_id: {}, "$other": {"error": "bad"}}},
+            )
+        )
+
+        self._send([self.event])
+
+        callback.assert_awaited_once_with(
+            FederationEventDeliveryEvent(
+                server_name=DEST,
+                events=[self.event],
+                method=FederatedEventDeliveryMethod.SEND,
+            )
+        )
+
+    def test_module_callback_rejected(self) -> None:
+        callback = AsyncMock(return_value=None)
+        self.hs.get_module_api_callbacks().federation.register_callbacks(
+            on_event_delivered_over_federation=callback
+        )
+        self.transport.responses.append(
+            json_response(200, {"pdus": {self.event.event_id: {"error": "bad"}}})
+        )
+
+        self._send([self.event])
+
+        # No PDU was accepted, so there is nothing to tell the module.
+        callback.assert_not_awaited()
+
+    def test_python_decoder_fallback(self) -> None:
+        """A body serde_json refuses, but Python's decoder accepts, still
+        reaches the module callback."""
+        callback = AsyncMock(return_value=None)
+        self.hs.get_module_api_callbacks().federation.register_callbacks(
+            on_event_delivered_over_federation=callback
+        )
+        body = (
+            '{"pdus": {"%s": {}, "$x": {"error": "\\ud800"}}}' % self.event.event_id
+        ).encode()
+        self.transport.responses.append(
+            (200, [(b"content-type", b"application/json")], body)
+        )
+
+        self._send([self.event])
+
+        callback.assert_awaited_once_with(
+            FederationEventDeliveryEvent(
+                server_name=DEST,
+                events=[self.event],
+                method=FederatedEventDeliveryMethod.SEND,
+            )
+        )
+
+    def test_last_sent_pdu_time(self) -> None:
+        self.transport.responses.append(json_response(200, {}))
+        self._send([self.event])
+
+        gauge = last_pdu_ts_metric.labels(
+            destination_server_name=DEST, **{SERVER_NAME_LABEL: self.hs.hostname}
+        )
+        self.assertEqual(
+            gauge._value.get(),
+            self.event.origin_server_ts / 1000,
+        )
+
+    def test_to_self(self) -> None:
+        self.get_failure(
+            self.manager.send_new_transaction(self.hs.hostname, [self.event], []),
+            RuntimeError,
+        )
+        self.assertEqual(self.transport.requests, [])
+
+    def test_gil_crossings(self) -> None:
+        """Below the facade, a transaction takes the GIL once to hand back the
+        result, plus once for the test-only body check."""
+        self.transport.responses.append(json_response(200, {}))
+        before = federation_gil_stats()
+
+        self._send([self.event])
+
+        after = federation_gil_stats()
+        calls = {
+            site: after[site][0] - before.get(site, (0, 0.0))[0]
+            for site in after
+            if after[site][0] != before.get(site, (0, 0.0))[0]
+        }
+        self.assertEqual(calls, {"transaction_check": 1, "transaction_response": 1})
+
+
+class RustTransactionsLevelTestCase(unittest.HomeserverTestCase):
+    """Which transaction sender each level gets."""
+
+    def _manager(self) -> TransactionManager:
+        return TransactionManager(self.hs)
+
+    @unittest.override_config(
+        {"experimental_features": {"rust_federation_sending": "client"}}
+    )
+    def test_client_level(self) -> None:
+        self.assertIsNone(self._manager()._rust)
+
+    @unittest.override_config(
+        {"experimental_features": {"rust_federation_sending": "transactions"}}
+    )
+    def test_transactions_level(self) -> None:
+        self.assertIsNotNone(self._manager()._rust)
+
+    @unittest.override_config(
+        {
+            "experimental_features": {"rust_federation_sending": "transactions"},
+            "instance_map": {
+                "main": {"host": "testserv", "port": 8765},
+                "federation_sender": {"host": "testserv", "port": 1001},
+            },
+            "outbound_federation_restricted_to": ["federation_sender"],
+            "worker_replication_secret": "secret",
+        }
+    )
+    def test_proxy_worker(self) -> None:
+        """An instance that sends through a proxy worker has no Rust client,
+        so it builds transactions in Python."""
+        self.assertIsNone(self._manager()._rust)

tests/server.py

@@ -89,6 +89,7 @@ from synapse.config.database import DatabaseConnectionConfig
 from synapse.config.homeserver import HomeServerConfig
 from synapse.events.auto_accept_invites import InviteAutoAccepter
 from synapse.events.presence_router import load_legacy_presence_router
+from synapse.federation.sender.transaction_manager import TransactionManager
 from synapse.handlers.auth import load_legacy_password_auth_providers
 from synapse.http.site import SynapseRequest
 from synapse.logging.context import ContextResourceUsage
@@ -123,6 +124,7 @@ from tests.utils import (
     POSTGRES_PASSWORD,
     POSTGRES_PORT,
     POSTGRES_USER,
+    RUST_FED_STRICT_FOR_TESTS,
     RUST_STATE_RES_FOR_TESTS,
     RUST_SYNC_FOR_TESTS,
     RUST_SYNC_STRICT_FOR_TESTS,
@@ -134,6 +136,8 @@ from tests.utils import (
 
 logger = logging.getLogger(__name__)
 
+TransactionManager._rust_strict = RUST_FED_STRICT_FOR_TESTS
+
 R = TypeVar("R")
 P = ParamSpec("P")
 

tests/utils.py

@@ -64,6 +64,9 @@ RUST_SYNC_FOR_TESTS = os.environ.get("SYNAPSE_TEST_RUST_SYNC")
 # Python one, fails the test, so that the suite is a differential test of the
 # two. Set to 0 to only count and log them, as production does.
 RUST_SYNC_STRICT_FOR_TESTS = os.environ.get("SYNAPSE_TEST_RUST_SYNC_STRICT", "1") != "0"
+# At `rust_federation_sending: transactions`, check every transaction body Rust
+# builds against the one Python would build, and fail the test if they differ.
+RUST_FED_STRICT_FOR_TESTS = os.environ.get("SYNAPSE_TEST_RUST_FED_STRICT", "1") != "0"
 LEAVE_DB = os.environ.get("SYNAPSE_LEAVE_DB", False)
 POSTGRES_USER = os.environ.get("SYNAPSE_POSTGRES_USER", None)
 POSTGRES_HOST = os.environ.get("SYNAPSE_POSTGRES_HOST", None)