Send federation requests through Rust at rust_federation_sending: client

Add the `experimental_features.rust_federation_sending` level, `off` or
`client` for now. At `client`, `MatrixFederationHttpClient` sends its
JSON requests (`put_json`, `post_json`, `get_json`,
`get_json_with_headers`, `delete_json`, `send_direct_request`) through
the Rust client. Media downloads, and an instance that sends through a
proxy worker, stay on the Python path.

The facade keeps the tracing span around the call and injects its
headers. The request and response counters are bumped from the GIL hop
that returns the result. The limiter's background write is factored out
of `RetryDestinationLimiter` so both paths share it.

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

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.

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
+

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,

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)

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