From 5aff3307b253cb95cc4cda77aad0fa465ff559ba Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 15 Oct 2026 13:37:32 +0000 Subject: [PATCH] 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 Claude-Session: https://claude.ai/code/session_01R5b7Ruy4cgt27Mjq6Ptxbb --- changelog.d/rust-fed-txn.misc | 1 + synapse/config/experimental.py | 28 +- .../federation/sender/transaction_manager.py | 101 ++++++ synapse/http/matrixfederationclient.py | 2 +- tests/federation/_rust_transactions.py | 99 ++++++ tests/federation/test_federation_catch_up.py | 12 + tests/federation/test_federation_sender.py | 22 ++ tests/http/test_rust_transactions.py | 291 ++++++++++++++++++ tests/server.py | 4 + tests/utils.py | 3 + 10 files changed, 559 insertions(+), 4 deletions(-) create mode 100644 changelog.d/rust-fed-txn.misc create mode 100644 tests/federation/_rust_transactions.py create mode 100644 tests/http/test_rust_transactions.py diff --git a/changelog.d/rust-fed-txn.misc b/changelog.d/rust-fed-txn.misc new file mode 100644 index 0000000000..44d82a7946 --- /dev/null +++ b/changelog.d/rust-fed-txn.misc @@ -0,0 +1 @@ +Build and send federation transactions in Rust at `experimental_features.rust_federation_sending: transactions`. diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py index b3c3ad37a8..a03a834ef2 100644 --- a/synapse/config/experimental.py +++ b/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) diff --git a/synapse/federation/sender/transaction_manager.py b/synapse/federation/sender/transaction_manager.py index 9a7fb8a3b1..0314a035d1 100644 --- a/synapse/federation/sender/transaction_manager.py +++ b/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}" + ) diff --git a/synapse/http/matrixfederationclient.py b/synapse/http/matrixfederationclient.py index eb7f92e6ab..c2a9d30325 100644 --- a/synapse/http/matrixfederationclient.py +++ b/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( diff --git a/tests/federation/_rust_transactions.py b/tests/federation/_rust_transactions.py new file mode 100644 index 0000000000..5c07025521 --- /dev/null +++ b/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: +# . +# +"""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 + ) diff --git a/tests/federation/test_federation_catch_up.py b/tests/federation/test_federation_catch_up.py index 95a4fa57c1..462fe84c3b 100644 --- a/tests/federation/test_federation_catch_up.py +++ b/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.""" diff --git a/tests/federation/test_federation_sender.py b/tests/federation/test_federation_sender.py index 2a9c2f0fc4..e3a7be1c49 100644 --- a/tests/federation/test_federation_sender.py +++ b/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.""" diff --git a/tests/http/test_rust_transactions.py b/tests/http/test_rust_transactions.py new file mode 100644 index 0000000000..c78b9d2083 --- /dev/null +++ b/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: +# . +# +"""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/` 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) diff --git a/tests/server.py b/tests/server.py index 45439187cb..cc1ca4fc49 100644 --- a/tests/server.py +++ b/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") diff --git a/tests/utils.py b/tests/utils.py index 42f53c605b..a22da02476 100644 --- a/tests/utils.py +++ b/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)