diff --git a/changelog.d/rust-fed-client.misc b/changelog.d/rust-fed-client.misc new file mode 100644 index 0000000000..0f784761a8 --- /dev/null +++ b/changelog.d/rust-fed-client.misc @@ -0,0 +1 @@ +Add an experimental option, `rust_federation_sending`, to send federation requests through a Rust implementation of the request layer. diff --git a/synapse/config/experimental.py b/synapse/config/experimental.py index 9df38b0f1b..b3c3ad37a8 100644 --- a/synapse/config/experimental.py +++ b/synapse/config/experimental.py @@ -356,3 +356,20 @@ class ExperimentalConfig(Config): ("experimental", "rust_sync"), ) self.rust_sync: Literal["off", "shadow", "on"] = rust_sync + + # How much of outbound federation runs in Rust. One ordered level, each + # implying the ones before it: + # - "off": all Python. + # - "client": `MatrixFederationHttpClient` sends its JSON requests + # 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. + rust_federation_sending = experimental.get("rust_federation_sending", "off") + if rust_federation_sending not in ("off", "client"): + raise ConfigError( + "experimental_features.rust_federation_sending must be one of " + "'off' or 'client'", + ("experimental", "rust_federation_sending"), + ) + self.rust_federation_sending: Literal["off", "client"] = rust_federation_sending + diff --git a/synapse/http/matrixfederationclient.py b/synapse/http/matrixfederationclient.py index 61b38529b7..c721a1cf36 100644 --- a/synapse/http/matrixfederationclient.py +++ b/synapse/http/matrixfederationclient.py @@ -84,11 +84,13 @@ from synapse.logging import opentracing from synapse.logging.context import make_deferred_yieldable, run_in_background from synapse.logging.opentracing import set_tag, start_active_span, tags from synapse.metrics import SERVER_NAME_LABEL +from synapse.synapse_rust.federation import RustFederationClient from synapse.types import JsonDict from synapse.util.async_helpers import AwakenableSleeper, Linearizer, timeout_deferred from synapse.util.clock import Clock from synapse.util.json import json_decoder from synapse.util.metrics import Measure +from synapse.util.retryutils import store_retry_timings_in_background from synapse.util.stringutils import parse_and_validate_server_name if TYPE_CHECKING: @@ -489,13 +491,140 @@ class MatrixFederationHttpClient: ) self._is_shutdown = False + # At `rust_federation_sending: client` and above, the JSON requests go + # through the Rust request layer. Media downloads stay here, and so + # does an instance that sends through a proxy worker, since the Rust + # client only talks to federation directly. + self._rust_client: RustFederationClient | None = None + if ( + hs.config.experimental.rust_federation_sending != "off" + and hs.get_instance_name() in outbound_federation_restricted_to + ): + self._rust_client = RustFederationClient( + hs, + hs.version_string, + user_agent, + self._write_retry_timings, + self._record_request_metrics, + ) + def shutdown(self) -> None: self._is_shutdown = True + if self._rust_client is not None: + self._rust_client.shutdown() def wake_destination(self, destination: str) -> None: """Called when the remote server may have come back online.""" self._sleeper.wake(destination) + if self._rust_client is not None: + self._rust_client.wake_destination(destination) + + def _write_retry_timings( + self, + destination: str, + failure_ts: int | None, + retry_last_ts: int, + retry_interval: int, + server_up: bool, + ) -> None: + """Called by the Rust client when a destination's backoff changes.""" + store_retry_timings_in_background( + hs=self.hs, + store=self._store, + destination=destination, + failure_ts=failure_ts, + retry_last_ts=retry_last_ts, + retry_interval=retry_interval, + server_up=server_up, + notifier=self.hs.get_notifier(), + replication_client=self.hs.get_replication_command_handler(), + ) + + def _record_request_metrics( + self, method: str, attempts: int, response_codes: list[int] + ) -> None: + """Called by the Rust client once per request, with the number of + attempts it made and the status of each response.""" + if attempts: + outgoing_requests_counter.labels( + method=method, **{SERVER_NAME_LABEL: self.server_name} + ).inc(attempts) + for code in response_codes: + incoming_responses_counter.labels( + method=method, code=code, **{SERVER_NAME_LABEL: self.server_name} + ).inc() + + async def _send_rust_request( + self, + rust_client: RustFederationClient, + method: str, + destination: str, + path: str, + *, + args: QueryParams | None = None, + data: JsonDict | None = None, + parser: ByteParser[Any] | None = None, + want_headers: bool = False, + rewrite_age_ts: bool = False, + retry_on_dns_fail: bool = True, + timeout: int | None = None, + long_retries: bool = False, + ignore_backoff: bool = False, + backoff_on_404: bool = False, + backoff_on_all_error_codes: bool = False, + try_trailing_slash_on_400: bool = False, + ) -> tuple[Any, dict[bytes, list[bytes]] | None]: + """Send a request through the Rust client, which signs it, retries it, + applies the backoff and parses the response, as `_send_request` and + `_handle_response` do here. Returns the parsed body, and the response + headers if `want_headers`.""" + if parser is None or type(parser) is JsonParser: + parser_kind = "json" + parser = None + elif type(parser) is LegacyJsonSendParser: + parser_kind = "legacy_json_send" + parser = None + else: + parser_kind = "custom" + + with start_active_span( + "outgoing-federation-request", + tags={ + tags.SPAN_KIND: tags.SPAN_KIND_RPC_CLIENT, + tags.PEER_ADDRESS: destination, + tags.HTTP_METHOD: method, + tags.HTTP_URL: path, + }, + finish_on_close=True, + ): + 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 + ] + + return await rust_client.request( + method, + destination, + path, + encode_query_args(args).decode("ascii"), + data, + parser_kind, + parser, + rewrite_age_ts=rewrite_age_ts, + extra_headers=extra_headers, + retry_on_dns_fail=retry_on_dns_fail, + timeout_ms=timeout, + long_retries=long_retries, + ignore_backoff=ignore_backoff, + backoff_on_404=backoff_on_404, + backoff_on_all_error_codes=backoff_on_all_error_codes, + try_trailing_slash_on_400=try_trailing_slash_on_400, + want_headers=want_headers, + ) async def _send_request_with_optional_trailing_slash( self, @@ -1063,6 +1192,35 @@ class MatrixFederationHttpClient: RequestSendFailed: If there were problems connecting to the remote, due to e.g. DNS failures, connection timeouts etc. """ + if self._rust_client is not None: + # The only `json_data_callback` is the transaction manager's, + # which rewrites each PDU's `age_ts` to an `age` on every attempt, + # and is given `data` as well. The Rust client does that rewrite + # itself, so the callback is not called once per attempt. + rewrite_age_ts = False + if json_data_callback is not None: + if data is None: + data = json_data_callback() + else: + rewrite_age_ts = True + body, _ = await self._send_rust_request( + self._rust_client, + "PUT", + destination, + path, + args=args, + data=data, + parser=parser, + rewrite_age_ts=rewrite_age_ts, + long_retries=long_retries, + timeout=timeout, + ignore_backoff=ignore_backoff, + backoff_on_404=backoff_on_404, + try_trailing_slash_on_400=try_trailing_slash_on_400, + backoff_on_all_error_codes=backoff_on_all_error_codes, + ) + return body + request = MatrixFederationRequest( method="PUT", destination=destination, @@ -1152,6 +1310,19 @@ class MatrixFederationHttpClient: RequestSendFailed: If there were problems connecting to the remote, due to e.g. DNS failures, connection timeouts etc. """ + if self._rust_client is not None: + body, _ = await self._send_rust_request( + self._rust_client, + "POST", + destination, + path, + args=args, + data=data, + long_retries=long_retries, + timeout=timeout, + ignore_backoff=ignore_backoff, + ) + return body request = MatrixFederationRequest( method="POST", destination=destination, path=path, query=args, json=data @@ -1264,6 +1435,22 @@ class MatrixFederationHttpClient: RequestSendFailed: If there were problems connecting to the remote, due to e.g. DNS failures, connection timeouts etc. """ + if self._rust_client is not None: + # As below, without converting headers we would drop. + body, _ = await self._send_rust_request( + self._rust_client, + "GET", + destination, + path, + args=args, + parser=parser, + retry_on_dns_fail=retry_on_dns_fail, + timeout=timeout, + ignore_backoff=ignore_backoff, + try_trailing_slash_on_400=try_trailing_slash_on_400, + ) + return body + json_dict, _ = await self.get_json_with_headers( destination=destination, path=path, @@ -1358,6 +1545,23 @@ class MatrixFederationHttpClient: RequestSendFailed: If there were problems connecting to the remote, due to e.g. DNS failures, connection timeouts etc. """ + if self._rust_client is not None: + body, headers = await self._send_rust_request( + self._rust_client, + "GET", + destination, + path, + args=args, + parser=parser, + want_headers=True, + retry_on_dns_fail=retry_on_dns_fail, + timeout=timeout, + ignore_backoff=ignore_backoff, + try_trailing_slash_on_400=try_trailing_slash_on_400, + ) + assert headers is not None + return body, headers + request = MatrixFederationRequest( method="GET", destination=destination, path=path, query=args ) @@ -1439,6 +1643,19 @@ class MatrixFederationHttpClient: RequestSendFailed: If there were problems connecting to the remote, due to e.g. DNS failures, connection timeouts etc. """ + if self._rust_client is not None: + body, _ = await self._send_rust_request( + self._rust_client, + "DELETE", + destination, + path, + args=args, + long_retries=long_retries, + timeout=timeout, + ignore_backoff=ignore_backoff, + ) + return body + request = MatrixFederationRequest( method="DELETE", destination=destination, path=path, query=args ) @@ -1508,6 +1725,24 @@ class MatrixFederationHttpClient: RequestSendFailed: If there were problems connecting to the remote, due to e.g. DNS failures, connection timeouts etc. """ + if self._rust_client is not None: + try: + body, _ = await self._send_rust_request( + self._rust_client, + method, + destination, + path, + args=args, + data=json_body, + ignore_backoff=True, + timeout=timeout, + ) + except RequestSendFailed as e: + if isinstance(e.inner_exception, HttpResponseException): + raise e.inner_exception from e + raise + return body + request = MatrixFederationRequest( method=method, destination=destination, diff --git a/synapse/util/retryutils.py b/synapse/util/retryutils.py index 8a5aab50f1..89b63b728c 100644 --- a/synapse/util/retryutils.py +++ b/synapse/util/retryutils.py @@ -313,27 +313,55 @@ class RetryDestinationLimiter: # Whether the current request to the destination had been failing. currently_failing = bool(self.failure_ts) - async def store_retry_timings() -> None: - try: - await self.store.set_destination_retry_timings( - self.destination, - self.failure_ts, - retry_last_ts, - self.retry_interval, - ) + store_retry_timings_in_background( + hs=self.hs, + store=self.store, + destination=self.destination, + failure_ts=self.failure_ts, + retry_last_ts=retry_last_ts, + retry_interval=self.retry_interval, + server_up=previously_failing and not currently_failing, + notifier=self.notifier, + replication_client=self.replication_client, + ) - # If the server was previously failing, but is no longer. - if previously_failing and not currently_failing: - if self.notifier: - # Inform the relevant places that the remote server is back up. - self.notifier.notify_remote_server_up(self.destination) - if self.replication_client: - # Inform other workers that the remote server is up. - self.replication_client.send_remote_server_up(self.destination) +def store_retry_timings_in_background( + *, + hs: "HomeServer", + store: DataStore, + destination: str, + failure_ts: int | None, + retry_last_ts: int, + retry_interval: int, + server_up: bool, + notifier: Optional["Notifier"], + replication_client: Optional["ReplicationCommandHandler"], +) -> None: + """Write a destination's new retry timings, in the background. - except Exception: - logger.exception("Failed to store destination_retry_timings") + Args: + server_up: the destination was failing and now is not, so tell the + notifier and the other workers. + """ - # we deliberately do this in the background. - self.hs.run_as_background_process("store_retry_timings", store_retry_timings) + async def store_retry_timings() -> None: + try: + await store.set_destination_retry_timings( + destination, failure_ts, retry_last_ts, retry_interval + ) + + if server_up: + if notifier: + # Inform the relevant places that the remote server is back up. + notifier.notify_remote_server_up(destination) + + if replication_client: + # Inform other workers that the remote server is up. + replication_client.send_remote_server_up(destination) + + except Exception: + logger.exception("Failed to store destination_retry_timings") + + # we deliberately do this in the background. + hs.run_as_background_process("store_retry_timings", store_retry_timings) diff --git a/tests/http/test_rust_federation_client.py b/tests/http/test_rust_federation_client.py new file mode 100644 index 0000000000..9efa51814b --- /dev/null +++ b/tests/http/test_rust_federation_client.py @@ -0,0 +1,381 @@ +# +# 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 `MatrixFederationHttpClient` at `rust_federation_sending: client`, +where its JSON requests go through the Rust request layer. + +The Rust client sends through a Python callable here, in place of the network. +It runs on a tokio thread, so the tests wait for it with `get_success`, and +advance the reactor to get past the sleeps between retries. +""" + +import json +import re +import time +from collections import deque +from typing import Any + +import attr +from signedjson.key import get_verify_key +from signedjson.sign import verify_signed_json + +from twisted.internet.defer import Deferred, ensureDeferred +from twisted.internet.testing import MemoryReactor + +from synapse.api.errors import ( + FederationDeniedError, + HttpResponseException, + RequestSendFailed, +) +from synapse.http.matrixfederationclient import ByteParser, LegacyJsonSendParser +from synapse.server import HomeServer +from synapse.synapse_rust.federation import ( + RustFederationClient, + federation_gil_stats, +) +from synapse.types import JsonDict +from synapse.util.clock import Clock +from synapse.util.retryutils import NotRetryingDestination + +from tests import unittest + +DEST = "remote.example" + + +@attr.s(auto_attribs=True, frozen=True) +class SentRequest: + destination: str + method: str + path: str + headers: list[tuple[bytes, bytes]] + body: bytes + + def header(self, name: bytes) -> bytes | None: + for key, value in self.headers: + if key == name.lower(): + return value + return None + + +Response = tuple[int, list[tuple[bytes, bytes]], bytes] + + +def json_response(status: int, body: Any) -> Response: + return status, [(b"content-type", b"application/json")], json.dumps(body).encode() + + +class FakeTransport: + """Answers each request with the next queued response.""" + + def __init__(self) -> None: + self.responses: deque[Response] = deque() + self.requests: list[SentRequest] = [] + + def __call__( + self, + destination: str, + method: str, + path: str, + headers: list[tuple[bytes, bytes]], + body: bytes, + ) -> Response: + self.requests.append(SentRequest(destination, method, path, headers, body)) + if not self.responses: + raise AssertionError(f"unexpected request {method} {path}") + return self.responses.popleft() + + +class RustFederationClientTestCase(unittest.HomeserverTestCase): + def default_config(self) -> dict[str, Any]: + config = super().default_config() + config.setdefault("experimental_features", {})["rust_federation_sending"] = ( + "client" + ) + return config + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.cl = hs.get_federation_http_client() + assert self.cl._rust_client is not None + + self.transport = FakeTransport() + self.cl._rust_client = RustFederationClient( + hs, + hs.version_string, + hs.version_string, + self.cl._write_retry_timings, + self.cl._record_request_metrics, + self.transport, + ) + + def _wait(self, d: "Deferred[Any]", step: float = 0.5, limit: int = 4000) -> None: + """Advance the reactor in `step` second steps, giving tokio a moment + after each, until `d` fires.""" + for _ in range(limit): + if d.called and not d.paused: + return + self.reactor.advance(step) + time.sleep(0.001) + raise AssertionError("request did not finish") + + def test_get_json(self) -> None: + self.transport.responses.append(json_response(200, {"foo": "bar"})) + + result = self.get_success( + self.cl.get_json(DEST, "/_matrix/federation/v1/x", args={"a": ["1", "2"]}) + ) + + self.assertEqual(result, {"foo": "bar"}) + (request,) = self.transport.requests + self.assertEqual(request.destination, DEST) + self.assertEqual(request.method, "GET") + self.assertEqual(request.path, "/_matrix/federation/v1/x?a=1&a=2") + self.assertEqual(request.body, b"") + self.assertIsNone(request.header(b"Content-Type")) + self.assertEqual(request.header(b"User-Agent"), self.hs.version_string.encode()) + self._assert_signed(request, None) + + def test_put_json_body(self) -> None: + self.transport.responses.append(json_response(200, {})) + + self.get_success( + self.cl.put_json(DEST, "/_matrix/federation/v1/y", data={"b": 2, "a": "é"}) + ) + + (request,) = self.transport.requests + self.assertEqual(request.method, "PUT") + self.assertEqual(request.header(b"Content-Type"), b"application/json") + self.assertEqual(request.body, '{"a":"é","b":2}'.encode()) + self._assert_signed(request, {"b": 2, "a": "é"}) + + def _assert_signed(self, request: SentRequest, content: JsonDict | None) -> None: + """Check the `X-Matrix` header, as the receiving server would.""" + header = request.header(b"Authorization") + assert header is not None + params = dict(re.findall(r'(\w+)="([^"]*)"', header.decode())) + self.assertEqual(params["origin"], self.hs.hostname) + self.assertEqual(params["destination"], DEST) + + signed: JsonDict = { + "method": request.method, + "uri": request.path, + "origin": params["origin"], + "destination": params["destination"], + "signatures": {params["origin"]: {params["key"]: params["sig"]}}, + } + if content is not None: + signed["content"] = content + verify_key = get_verify_key(self.hs.signing_key) + verify_signed_json(signed, params["origin"], verify_key) + + def test_retries_server_errors(self) -> None: + self.transport.responses.extend( + [json_response(502, {}), json_response(500, {}), json_response(200, {})] + ) + + d = ensureDeferred(self.cl.get_json(DEST, "/x")) + self._wait(d) + + self.assertEqual(self.successResultOf(d), {}) + self.assertEqual(len(self.transport.requests), 3) + + def test_client_error_raises(self) -> None: + self.transport.responses.append( + json_response(404, {"errcode": "M_NOT_FOUND", "error": "nope"}) + ) + + failure = self.get_failure(self.cl.get_json(DEST, "/x"), HttpResponseException) + + self.assertEqual(failure.value.code, 404) + self.assertEqual(failure.value.to_synapse_error().errcode, "M_NOT_FOUND") + self.assertEqual(len(self.transport.requests), 1) + + def test_failures_start_a_backoff(self) -> None: + """Running out of retries records a backoff, which the next request + honours. This goes through the Python write and its cache + invalidation, which reaches the Rust client's copy of the timings.""" + # The first attempt plus the three short retries. + self.transport.responses.extend([json_response(500, {})] * 4) + + d = ensureDeferred(self.cl.get_json(DEST, "/x")) + self._wait(d) + failure = self.failureResultOf(d, RequestSendFailed) + assert isinstance(failure.value, RequestSendFailed) + self.assertTrue(failure.value.can_retry) + self.assertIsInstance(failure.value.inner_exception, HttpResponseException) + + timings = self.get_success( + self.hs.get_datastores().main.get_destination_retry_timings(DEST) + ) + assert timings is not None + self.assertEqual( + timings.retry_interval, + self.hs.config.federation.destination_min_retry_interval_ms, + ) + + self.get_failure(self.cl.get_json(DEST, "/x"), NotRetryingDestination) + self.assertEqual(len(self.transport.requests), 4) + + # Ignoring the backoff sends, and success clears it. + self.transport.responses.append(json_response(200, {})) + self.get_success(self.cl.get_json(DEST, "/x", ignore_backoff=True)) + self.reactor.advance(0) + timings = self.get_success( + self.hs.get_datastores().main.get_destination_retry_timings(DEST) + ) + self.assertIsNone(timings) + + def test_send_direct_request_raises_the_http_error(self) -> None: + self.transport.responses.extend([json_response(503, {})] * 4) + + d = ensureDeferred(self.cl.send_direct_request("GET", DEST, "/x")) + self._wait(d) + + failure = self.failureResultOf(d, HttpResponseException) + self.assertEqual(failure.value.code, 503) + + def test_transport_error_with_a_timeout_is_not_retried(self) -> None: + def fail(*args: Any) -> Response: + raise ConnectionError("boom") + + self.cl._rust_client = RustFederationClient( + self.hs, + self.hs.version_string, + self.hs.version_string, + self.cl._write_retry_timings, + self.cl._record_request_metrics, + fail, + ) + d = ensureDeferred(self.cl.get_json(DEST, "/x", timeout=10_000)) + self._wait(d) + failure = self.failureResultOf(d, RequestSendFailed) + assert isinstance(failure.value, RequestSendFailed) + self.assertIsInstance(failure.value.inner_exception, ConnectionError) + self.assertTrue(failure.value.can_retry) + + @unittest.override_config({"federation_domain_whitelist": ["other.example"]}) + def test_whitelist(self) -> None: + self.get_failure(self.cl.get_json(DEST, "/x"), FederationDeniedError) + self.assertEqual(self.transport.requests, []) + + def test_invalid_json(self) -> None: + self.transport.responses.append( + (200, [(b"content-type", b"application/json")], b"{not json") + ) + failure = self.get_failure(self.cl.get_json(DEST, "/x"), RequestSendFailed) + self.assertFalse(failure.value.can_retry) + self.assertIsInstance(failure.value.inner_exception, ValueError) + + def test_json_that_is_not_an_object(self) -> None: + self.transport.responses.append(json_response(200, [1, 2])) + failure = self.get_failure(self.cl.get_json(DEST, "/x"), RequestSendFailed) + self.assertFalse(failure.value.can_retry) + self.assertIn("list", str(failure.value.inner_exception)) + + def test_json_python_accepts_and_serde_does_not(self) -> None: + """A lone surrogate is valid JSON to Python.""" + self.transport.responses.append( + (200, [(b"content-type", b"application/json")], b'{"a": "\\ud800"}') + ) + result = self.get_success(self.cl.get_json(DEST, "/x")) + self.assertEqual(result, {"a": "\ud800"}) + + def test_wrong_content_type(self) -> None: + self.transport.responses.append((200, [(b"content-type", b"text/html")], b"{}")) + failure = self.get_failure(self.cl.get_json(DEST, "/x"), RequestSendFailed) + self.assertFalse(failure.value.can_retry) + self.assertIn("text/html", str(failure.value.inner_exception)) + + def test_legacy_json_send_parser(self) -> None: + self.transport.responses.append(json_response(200, [200, {"ok": True}])) + result = self.get_success( + self.cl.put_json( + DEST, "/send_join", data={"a": 1}, parser=LegacyJsonSendParser() + ) + ) + self.assertEqual(result, [200, {"ok": True}]) + + self.transport.responses.append(json_response(200, {"ok": True})) + self.get_failure( + self.cl.put_json( + DEST, "/send_join", data={"a": 1}, parser=LegacyJsonSendParser() + ), + RequestSendFailed, + ) + + def test_custom_parser(self) -> None: + class Collect(ByteParser[bytes]): + CONTENT_TYPE = "application/json" + + def __init__(self) -> None: + self.data = b"" + + def write(self, data: bytes) -> int: + self.data += data + return len(data) + + def finish(self) -> bytes: + return self.data + + body = json.dumps({"big": "x" * 200_000}).encode() + self.transport.responses.append( + (200, [(b"content-type", b"application/json")], body) + ) + result = self.get_success(self.cl.get_json(DEST, "/x", parser=Collect())) + self.assertEqual(result, body) + + def test_get_json_with_headers(self) -> None: + self.transport.responses.append( + ( + 200, + [(b"content-type", b"application/json"), (b"x-thing", b"1")], + b"{}", + ) + ) + _, headers = self.get_success(self.cl.get_json_with_headers(DEST, "/x")) + self.assertEqual(headers[b"X-Thing"], [b"1"]) + self.assertEqual(headers[b"Content-Type"], [b"application/json"]) + + def test_transaction_age_rewrite(self) -> None: + """`send_transaction`'s `json_data_callback` is replaced by the Rust + client rewriting `age_ts` on each attempt.""" + self.transport.responses.append(json_response(200, {})) + now = self.clock.time_msec() + data = {"pdus": [{"event_id": "$a", "age_ts": now - 1000}]} + + def callback() -> JsonDict: + raise AssertionError("the callback is not called") + + self.get_success( + self.cl.put_json(DEST, "/send/1", data=data, json_data_callback=callback) + ) + (request,) = self.transport.requests + self.assertEqual( + json.loads(request.body)["pdus"], + [{"event_id": "$a", "unsigned": {"age": 1000}}], + ) + + def test_gil_crossings(self) -> None: + """A successful request takes the GIL once below the facade, to hand + back the result. The transport here is Python and is not counted.""" + self.transport.responses.append(json_response(200, {})) + before = federation_gil_stats() + + self.get_success(self.cl.get_json(DEST, "/x")) + + 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, {"response": 1})