Add Rust per-destination federation queues

`DestinationQueue` holds what waits to go to one destination: PDUs,
EDUs, keyed EDUs, presence and receipts, with Python's merge rules and
limits. The first thing queued starts its transmission loop as a tokio
task, which builds and sends transactions until there is nothing left,
as `_transaction_transmission_loop` does.

Catch-up, the sticky event backlog and the loop's database work stay
Python for now, behind `QueueHooks`. `RustDestinationQueue` is the
pyclass the Python facade wraps, and calls those hooks on it.

`run_python_awaitable_observed` converts a Python result on the reactor
and tells an observer when the awaitable starts and finishes. The loop
uses it so tests can tell a loop waiting on the reactor from one with
work to do.

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

rust/src/federation/queue/python.rs

@@ -0,0 +1,665 @@
+/*
+ * 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>.
+ *
+ */
+
+//! The queues as Python's `FederationSender` uses them.
+//!
+//! [`RustDestinationQueues`] makes a [`RustDestinationQueue`] for each
+//! destination. The Python `RustPerDestinationQueue` wraps one and provides
+//! the [`QueueHooks`] as coroutines: `_rust_catch_up`, `_rust_device_edus`,
+//! `_rust_sticky_backlog` and `_rust_complete`.
+
+use std::sync::{Arc, Mutex};
+
+use async_trait::async_trait;
+use pyo3::{
+    intern,
+    prelude::*,
+    types::{PyDict, PyList, PyString, PyTuple},
+};
+use serde_json::Value;
+
+use super::{
+    Activity, Completion, DestinationQueue, DeviceEdus, DeviceEdusRequest, Edu, LoopActivity,
+    LoopError, PduBody, PresenceState, QueueConfig, QueueHooks, QueuedPdu, Receipt, SendOutcome,
+    StickyBatch, StickyStreamPosition, TransactionSink,
+};
+use crate::{
+    deferred::{run_python_awaitable_observed, AwaitableObserver},
+    events::Event,
+    federation::{
+        client::{limiter::NotRetrying, RequestStats},
+        gil::{account, with_gil},
+        transaction::{
+            acknowledged_pdus,
+            python::{python_acknowledged, RustTransaction, RustTransactionManager},
+            Transaction, TransactionResponse, TransactionSender,
+        },
+    },
+    runtime::RustRuntime,
+};
+
+/// Makes the queues, and tells tests when their loops are idle.
+#[pyclass(frozen, name = "RustDestinationQueues")]
+pub struct RustDestinationQueues {
+    sink: Arc<ClientSink>,
+    runtime: RustRuntime,
+    activity: Arc<Activity>,
+    msc4354_enabled: bool,
+}
+
+#[pymethods]
+impl RustDestinationQueues {
+    #[new]
+    fn py_new(transactions: &Bound<'_, RustTransactionManager>, msc4354_enabled: bool) -> Self {
+        let transactions = transactions.get();
+        RustDestinationQueues {
+            sink: Arc::new(ClientSink {
+                sender: Arc::clone(&transactions.sender),
+                runtime: transactions.runtime.clone(),
+                record_metrics: Arc::clone(&transactions.record_metrics),
+                send_for_test: Mutex::new(None),
+            }),
+            runtime: transactions.runtime.clone(),
+            activity: Arc::default(),
+            msc4354_enabled,
+        }
+    }
+
+    /// The queue for `destination`. `facade` is the Python queue, which
+    /// provides the hooks. The Rust queue holds it weakly.
+    fn queue(
+        &self,
+        py: Python<'_>,
+        destination: String,
+        facade: &Bound<'_, PyAny>,
+        should_send_on_this_instance: bool,
+    ) -> PyResult<RustDestinationQueue> {
+        let facade = py
+            .import("weakref")?
+            .getattr("ref")?
+            .call1((facade,))?
+            .unbind();
+        let activity = LoopActivity::new(Arc::clone(&self.activity));
+        let hooks = PyHooks {
+            runtime: self.runtime.clone(),
+            facade: Arc::new(facade),
+            activity: Arc::clone(&activity),
+            destination: destination.clone(),
+        };
+        let sink: Arc<dyn TransactionSink> = self.sink.clone();
+        Ok(RustDestinationQueue {
+            queue: DestinationQueue::new(
+                QueueConfig {
+                    destination,
+                    should_send_on_this_instance,
+                    msc4354_enabled: self.msc4354_enabled,
+                },
+                Arc::new(hooks),
+                sink,
+                activity,
+                self.runtime.tokio_handle()?,
+            ),
+        })
+    }
+
+    /// Whether every loop has finished or is waiting on the reactor. For
+    /// tests, which pump the reactor until it is.
+    fn idle(&self) -> bool {
+        self.activity.idle()
+    }
+
+    /// Send transactions through `send(transaction)`, a coroutine function
+    /// taking a `RustTransaction` and returning the response dict, in place
+    /// of the network. For tests.
+    fn set_send_for_test(&self, send: Option<Py<PyAny>>) {
+        *self.sink.send_for_test.lock().expect("poisoned") = send.map(Arc::new);
+    }
+}
+
+/// One destination's queue. Each method is the `PerDestinationQueue` method
+/// of the same name.
+#[pyclass(frozen, name = "RustDestinationQueue")]
+pub struct RustDestinationQueue {
+    queue: Arc<DestinationQueue>,
+}
+
+#[pymethods]
+impl RustDestinationQueue {
+    fn shutdown(&self) {
+        self.queue.shutdown();
+    }
+
+    fn pending_pdu_count(&self) -> usize {
+        self.queue.pending_pdu_count()
+    }
+
+    fn pending_edu_count(&self) -> usize {
+        self.queue.pending_edu_count()
+    }
+
+    #[getter]
+    fn transmission_loop_running(&self) -> bool {
+        self.queue.transmission_loop_running()
+    }
+
+    #[getter]
+    fn new_data_to_send(&self) -> bool {
+        self.queue.new_data_to_send()
+    }
+
+    fn send_pdu(&self, pdu: &Bound<'_, Event>) {
+        self.queue.send_pdu(Arc::new(queued_pdu(pdu)));
+    }
+
+    #[pyo3(signature = (states, start_loop = true))]
+    fn send_presence(&self, states: Vec<Bound<'_, PyAny>>, start_loop: bool) -> PyResult<()> {
+        let states = states
+            .iter()
+            .map(presence_state)
+            .collect::<PyResult<Vec<_>>>()?;
+        self.queue.send_presence(states, start_loop);
+        Ok(())
+    }
+
+    fn queue_read_receipt(&self, receipt: &Bound<'_, PyAny>) -> PyResult<()> {
+        let py = receipt.py();
+        let data: Value = pythonize::depythonize(&receipt.getattr(intern!(py, "data"))?)?;
+        let Value::Object(data) = data else {
+            return Err(pyo3::exceptions::PyTypeError::new_err(
+                "receipt data must be a dict",
+            ));
+        };
+        self.queue.queue_read_receipt(&Receipt {
+            room_id: receipt.getattr(intern!(py, "room_id"))?.extract()?,
+            receipt_type: receipt.getattr(intern!(py, "receipt_type"))?.extract()?,
+            user_id: receipt.getattr(intern!(py, "user_id"))?.extract()?,
+            thread_id: receipt.getattr(intern!(py, "thread_id"))?.extract()?,
+            event_ids: receipt.getattr(intern!(py, "event_ids"))?.extract()?,
+            data,
+        });
+        Ok(())
+    }
+
+    /// `key` is the `repr` of Python's key, which is any hashable.
+    fn send_keyed_edu(&self, edu: &Bound<'_, PyAny>, key: String) -> PyResult<()> {
+        self.queue.send_keyed_edu(edu_from_py(edu)?, key);
+        Ok(())
+    }
+
+    fn send_edu(&self, edu: &Bound<'_, PyAny>) -> PyResult<()> {
+        self.queue.send_edu(edu_from_py(edu)?);
+        Ok(())
+    }
+
+    fn mark_new_data(&self) {
+        self.queue.mark_new_data();
+    }
+
+    fn attempt_new_transaction(&self) {
+        self.queue.attempt_new_transaction();
+    }
+
+    // The catch-up state, for the Python catch-up loop.
+
+    #[getter]
+    fn catching_up(&self) -> bool {
+        self.queue.catching_up()
+    }
+
+    #[setter]
+    fn set_catching_up(&self, catching_up: bool) {
+        self.queue.set_catching_up(catching_up);
+    }
+
+    #[getter]
+    fn catchup_last_skipped(&self) -> i64 {
+        self.queue.catchup_last_skipped()
+    }
+
+    #[setter]
+    fn set_catchup_last_skipped(&self, stream_ordering: i64) {
+        self.queue.set_catchup_last_skipped(stream_ordering);
+    }
+
+    #[getter]
+    fn last_successful_stream_ordering(&self) -> Option<i64> {
+        self.queue.last_successful_stream_ordering()
+    }
+
+    #[setter]
+    fn set_last_successful_stream_ordering(&self, stream_ordering: Option<i64>) {
+        self.queue
+            .set_last_successful_stream_ordering(stream_ordering);
+    }
+
+    #[getter]
+    fn transmission_loop_enabled(&self) -> bool {
+        self.queue.transmission_loop_enabled()
+    }
+
+    fn start_catching_up(&self) {
+        self.queue.start_catching_up();
+    }
+
+    #[getter]
+    fn sticky_backlogged(&self) -> bool {
+        self.queue.sticky_backlogged()
+    }
+
+    #[setter]
+    fn set_sticky_backlogged(&self, backlogged: bool) {
+        self.queue.set_sticky_backlogged(backlogged);
+    }
+
+    /// The event IDs of the queued PDUs, oldest first. For tests.
+    fn pending_pdu_ids(&self) -> Vec<String> {
+        self.queue.pending_pdu_ids()
+    }
+}
+
+/// A PDU for the queue. Its JSON is read later, on tokio.
+fn queued_pdu(pdu: &Bound<'_, Event>) -> QueuedPdu {
+    let event = pdu.get();
+    QueuedPdu {
+        event_id: event.event_id().to_owned(),
+        stream_ordering: event
+            .internal_metadata()
+            .stream_ordering()
+            .map(|so| so.get()),
+        origin_server_ts: event.origin_server_ts(),
+        body: PduBody::Event(pdu.clone().unbind()),
+    }
+}
+
+/// The parts of a `UserPresenceState` that go over federation.
+fn presence_state(state: &Bound<'_, PyAny>) -> PyResult<PresenceState> {
+    let py = state.py();
+    Ok(PresenceState {
+        user_id: state.getattr(intern!(py, "user_id"))?.extract()?,
+        state: state.getattr(intern!(py, "state"))?.extract()?,
+        last_active_ts: state.getattr(intern!(py, "last_active_ts"))?.extract()?,
+        status_msg: state.getattr(intern!(py, "status_msg"))?.extract()?,
+        currently_active: state
+            .getattr(intern!(py, "currently_active"))?
+            .is_truthy()?,
+    })
+}
+
+/// An `Edu`'s type and content.
+fn edu_from_py(edu: &Bound<'_, PyAny>) -> PyResult<Edu> {
+    let py = edu.py();
+    Ok(Edu {
+        edu_type: edu.getattr(intern!(py, "edu_type"))?.extract()?,
+        content: pythonize::depythonize(&edu.getattr(intern!(py, "content"))?)?,
+    })
+}
+
+/// The hooks, as coroutine methods of the Python queue.
+struct PyHooks {
+    runtime: RustRuntime,
+    /// A weak reference to the Python queue.
+    facade: Arc<Py<PyAny>>,
+    activity: Arc<LoopActivity>,
+    destination: String,
+}
+
+impl PyHooks {
+    /// Await `facade.<method>(*args(py))` on the reactor, and read its result
+    /// with `convert` there.
+    async fn call<A, C, T>(
+        &self,
+        site: &'static str,
+        method: &'static str,
+        args: A,
+        convert: C,
+    ) -> Result<T, LoopError>
+    where
+        A: for<'py> Fn(Python<'py>) -> PyResult<Bound<'py, PyTuple>> + Send + Sync + 'static,
+        C: for<'py> Fn(&Bound<'py, PyAny>) -> PyResult<T> + Send + Sync + 'static,
+        T: Send + 'static,
+    {
+        let facade = Arc::clone(&self.facade);
+        let observer: Arc<dyn AwaitableObserver> = self.activity.clone();
+        let result = run_python_awaitable_observed(
+            &self.runtime,
+            move |py| {
+                account(site, py, |_| {
+                    let facade = facade.bind(py).call0()?;
+                    if facade.is_none() {
+                        return Err(pyo3::exceptions::PyRuntimeError::new_err(
+                            "the federation queue has gone",
+                        ));
+                    }
+                    facade.call_method1(method, args(py)?)
+                })
+            },
+            convert,
+            Some(observer),
+        )
+        .await;
+        result.map_err(|err| python_error(err, &self.destination))
+    }
+}
+
+#[async_trait]
+impl QueueHooks for PyHooks {
+    async fn catch_up(&self) -> Result<(), LoopError> {
+        self.call(
+            "queue_catch_up",
+            "_rust_catch_up",
+            |py| Ok(PyTuple::empty(py)),
+            |_| Ok(()),
+        )
+        .await
+    }
+
+    async fn device_edus(&self, request: DeviceEdusRequest) -> Result<DeviceEdus, LoopError> {
+        self.call(
+            "queue_device_edus",
+            "_rust_device_edus",
+            move |py| {
+                PyTuple::new(
+                    py,
+                    [
+                        request.last_device_stream_id.into_pyobject(py)?,
+                        request.last_device_list_stream_id.into_pyobject(py)?,
+                        request.to_device_limit.into_pyobject(py)?,
+                        request.edu_limit.into_pyobject(py)?,
+                    ],
+                )
+            },
+            |result| {
+                let (to_device, to_device_stream_id, device_list, device_list_stream_id): (
+                    Vec<Bound<'_, PyAny>>,
+                    i64,
+                    Vec<(String, Bound<'_, PyAny>)>,
+                    i64,
+                ) = result.extract()?;
+                Ok(DeviceEdus {
+                    to_device: to_device
+                        .iter()
+                        .map(|content| pythonize::depythonize(content))
+                        .collect::<Result<_, _>>()?,
+                    to_device_stream_id,
+                    device_list: device_list
+                        .iter()
+                        .map(|(edu_type, content)| -> PyResult<Edu> {
+                            Ok(Edu {
+                                edu_type: edu_type.clone(),
+                                content: pythonize::depythonize(content)?,
+                            })
+                        })
+                        .collect::<PyResult<_>>()?,
+                    device_list_stream_id,
+                })
+            },
+        )
+        .await
+    }
+
+    async fn sticky_backlog(&self) -> Result<Option<StickyBatch>, LoopError> {
+        self.call(
+            "queue_sticky_backlog",
+            "_rust_sticky_backlog",
+            |py| Ok(PyTuple::empty(py)),
+            |result| {
+                if result.is_none() {
+                    return Ok(None);
+                }
+                let (room_id, position, events): (String, Py<PyAny>, Vec<Bound<'_, PyAny>>) =
+                    result.extract()?;
+                let pdus = events
+                    .iter()
+                    .map(|event| Ok(Arc::new(queued_pdu(event.cast::<Event>()?))))
+                    .collect::<PyResult<_>>()?;
+                let position: StickyStreamPosition = Arc::new(position);
+                Ok(Some(StickyBatch {
+                    room_id,
+                    position,
+                    pdus,
+                }))
+            },
+        )
+        .await
+    }
+
+    async fn complete(&self, completion: Completion) -> Result<(), LoopError> {
+        let completion = Arc::new(completion);
+        self.call(
+            "queue_complete",
+            "_rust_complete",
+            move |py| completion_args(py, &completion),
+            |_| Ok(()),
+        )
+        .await
+    }
+}
+
+/// The arguments of `_rust_complete`.
+fn completion_args<'py>(py: Python<'py>, completion: &Completion) -> PyResult<Bound<'py, PyTuple>> {
+    let sticky = match &completion.sticky {
+        Some((room_id, position)) => {
+            let position = position
+                .downcast_ref::<Py<PyAny>>()
+                .map(|position| position.clone_ref(py))
+                .unwrap_or_else(|| py.None());
+            (room_id.as_str(), position).into_pyobject(py)?.into_any()
+        }
+        None => py.None().into_bound(py),
+    };
+    let sent = match &completion.sent {
+        Some(sent) => {
+            let pdus = PyList::empty(py);
+            for pdu in &sent.pdus {
+                match &pdu.body {
+                    PduBody::Event(event) => pdus.append(event.clone_ref(py))?,
+                    #[cfg(test)]
+                    PduBody::Json(_) => {}
+                }
+            }
+            (
+                pdus,
+                &sent.edu_types,
+                &sent.acknowledged,
+                sent.stats.attempts,
+                &sent.stats.responses,
+            )
+                .into_pyobject(py)?
+                .into_any()
+        }
+        None => py.None().into_bound(py),
+    };
+    PyTuple::new(
+        py,
+        [
+            completion.to_device_stream_id.into_pyobject(py)?,
+            completion.device_list_stream_id.into_pyobject(py)?,
+            completion.last_stream_ordering.into_pyobject(py)?,
+            sticky,
+            sent,
+        ],
+    )
+}
+
+/// What a Python exception means to the loop. Takes the GIL, which only
+/// error paths need.
+fn python_error(err: PyErr, destination: &str) -> LoopError {
+    with_gil("queue_error", |py| {
+        classify_python_error(py, &err).unwrap_or_else(|classify_err| {
+            LoopError::Other(format!(
+                "{} (and classifying it failed: {classify_err}) for {destination}",
+                format_exception(py, &err)
+            ))
+        })
+    })
+}
+
+fn classify_python_error(py: Python<'_>, err: &PyErr) -> PyResult<LoopError> {
+    let value = err.value(py);
+    let errors = py.import("synapse.api.errors")?;
+    let not_retrying = py
+        .import("synapse.util.retryutils")?
+        .getattr("NotRetryingDestination")?;
+
+    Ok(if value.is_instance(&not_retrying)? {
+        LoopError::NotRetrying(NotRetrying {
+            retry_last_ts: value.getattr("retry_last_ts")?.extract()?,
+            retry_interval: value.getattr("retry_interval")?.extract()?,
+        })
+    } else if value.is_instance(&errors.getattr("FederationDeniedError")?)? {
+        LoopError::Denied(value.str()?.to_string())
+    } else if value.is_instance(&errors.getattr("HttpResponseException")?)? {
+        LoopError::Http {
+            code: value.getattr("code")?.extract()?,
+            message: value.str()?.to_string(),
+        }
+    } else if value.is_instance(&errors.getattr("RequestSendFailed")?)? {
+        LoopError::SendFailed(value.str()?.to_string())
+    } else {
+        LoopError::Other(format_exception(py, err))
+    })
+}
+
+/// The exception with its traceback, as `logger.exception` logs it.
+fn format_exception(py: Python<'_>, err: &PyErr) -> String {
+    let formatted = (|| -> PyResult<String> {
+        let lines = py
+            .import("traceback")?
+            .call_method1("format_exception", (err.value(py),))?;
+        Ok(PyString::new(py, "")
+            .call_method1("join", (lines,))?
+            .extract::<String>()?
+            .trim_end()
+            .to_owned())
+    })();
+    formatted.unwrap_or_else(|_| err.to_string())
+}
+
+/// Sends through the Rust transaction sender, or a Python stand-in in tests.
+struct ClientSink {
+    sender: Arc<TransactionSender>,
+    runtime: RustRuntime,
+    record_metrics: Arc<Py<PyAny>>,
+    send_for_test: Mutex<Option<Arc<Py<PyAny>>>>,
+}
+
+impl ClientSink {
+    /// Send through the Python stand-in.
+    async fn send_for_test(
+        &self,
+        send: Arc<Py<PyAny>>,
+        transaction: Transaction,
+        observer: Arc<dyn AwaitableObserver>,
+    ) -> Result<SendOutcome, LoopError> {
+        let destination = transaction.destination.clone();
+        let log_prefix = transaction.log_prefix();
+        let transaction = Arc::new(transaction);
+        let response = run_python_awaitable_observed(
+            &self.runtime,
+            move |py| {
+                let built = RustTransaction {
+                    inner: (*transaction).clone(),
+                };
+                send.bind(py).call1((built,))
+            },
+            |response| {
+                Ok(match response.cast::<PyDict>() {
+                    Ok(response) => pythonize::depythonize::<Value>(response)?,
+                    // A mock that returns nothing.
+                    Err(_) => Value::Null,
+                })
+            },
+            Some(observer),
+        )
+        .await
+        .map_err(|err| python_error(err, &destination))?;
+
+        let acknowledged = match response {
+            Value::Object(response) => acknowledged_pdus(&log_prefix, &response),
+            _ => Vec::new(),
+        };
+        Ok(SendOutcome {
+            acknowledged,
+            stats: RequestStats::default(),
+        })
+    }
+}
+
+#[async_trait]
+impl TransactionSink for ClientSink {
+    fn now_ms(&self) -> u64 {
+        self.sender.client().now_ms()
+    }
+
+    async fn backoff(&self, destination: &str) -> Result<Option<NotRetrying>, LoopError> {
+        self.sender
+            .client()
+            .backoff(destination)
+            .await
+            .map_err(|err| LoopError::Other(err.to_string()))
+    }
+
+    fn build(&self, destination: &str, pdus: Vec<Value>, edus: Vec<Value>) -> Transaction {
+        self.sender.build(destination, pdus, edus)
+    }
+
+    async fn send(
+        &self,
+        transaction: Transaction,
+        observer: Arc<dyn AwaitableObserver>,
+    ) -> Result<SendOutcome, LoopError> {
+        let send_for_test = self.send_for_test.lock().expect("poisoned").clone();
+        if let Some(send) = send_for_test {
+            return self.send_for_test(send, transaction, observer).await;
+        }
+
+        let destination = transaction.destination.clone();
+        let transaction_id = transaction.transaction_id.clone();
+        let mut stats = RequestStats::default();
+        let result = self.sender.send(transaction, Vec::new(), &mut stats).await;
+        match result {
+            Ok(TransactionResponse::Acknowledged(acknowledged)) => Ok(SendOutcome {
+                acknowledged,
+                stats,
+            }),
+            Ok(TransactionResponse::Unparsed(body)) => with_gil("queue_response_fallback", |py| {
+                python_acknowledged(py, &destination, &transaction_id, &body).map_err(|err| {
+                    classify_python_error(py, &err)
+                        .unwrap_or_else(|_| LoopError::Other(format_exception(py, &err)))
+                })
+            })
+            .map(|acknowledged| SendOutcome {
+                acknowledged,
+                stats,
+            }),
+            Err(err) => {
+                // The request counters are Python's. A failure is the only
+                // case without a later trip to Python to carry them.
+                let record_metrics = Arc::clone(&self.record_metrics);
+                with_gil("queue_send_failed", |py| {
+                    let recorded = record_metrics.bind(py).call1((
+                        "PUT",
+                        stats.attempts,
+                        stats.responses.clone(),
+                    ));
+                    if let Err(err) = recorded {
+                        log::warn!("Failed to record federation request metrics: {err}");
+                    }
+                });
+                Err(LoopError::from_client(err, &destination))
+            }
+        }
+    }
+}