Fix review findings in the Rust federation queues

- Stop catch-up before the next room once the loop is shut down. The Rust
  loop cannot cancel the Python catch-up coroutine, and after a shutdown
  the transaction manager returns without sending, so the room would be
  recorded as sent.
- Record the request metrics when the Python fallback for reading a
  transaction response fails.
- Give each transmission loop its own LoopActivity, so a loop that ends
  cannot clear the parked flag of the next.
- Count the time spent reading a hook's result as GIL time of its site.
- Correct the AwaitableObserver::finished doc.
- Let test_send_receipts_with_backoff_sender accept either order of its
  first two sends, which run at once with the queues in Rust.

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

rust/src/deferred.rs

@@ -244,7 +244,7 @@ pub trait AwaitableObserver: Send + Sync + 'static {
     /// The awaitable has been made and is running on the reactor.
     fn started(&self);
     /// The awaitable has a result, which the caller has yet to see. Also
-    /// called if it failed to start, without `started`.
+    /// called if it failed to start, whether or not `started` was.
     fn finished(&self);
 }
 

rust/src/federation/gil.rs

@@ -62,13 +62,24 @@ pub fn account<R>(site: &'static str, py: Python<'_>, f: impl FnOnce(Python<'_>)
     result
 }
 
+/// Add `held` to the time of `site` without counting a call: for later work
+/// in a GIL hold that [`account`] already counted, such as reading the result
+/// of a call it made.
+pub fn account_held(site: &'static str, held: Duration) {
+    add(site, 0, held);
+}
+
 fn record(site: &'static str, held: Duration) {
+    add(site, 1, held);
+}
+
+fn add(site: &'static str, calls: u64, held: Duration) {
     let mut stats = STATS.lock().expect("poisoned");
     let entry = stats
         .get_or_insert_with(HashMap::new)
         .entry(site)
         .or_default();
-    entry.calls += 1;
+    entry.calls += calls;
     entry.held += held;
 }
 
@@ -104,4 +115,14 @@ mod tests {
         assert_eq!(after.calls, before.calls + 2);
         assert!(after.held >= before.held);
     }
+
+    #[test]
+    fn held_time_without_a_call() {
+        let before = snapshot().get("held_site").copied().unwrap_or_default();
+        account_held("held_site", Duration::from_millis(5));
+        let after = snapshot()["held_site"];
+
+        assert_eq!(after.calls, before.calls);
+        assert_eq!(after.held, before.held + Duration::from_millis(5));
+    }
 }

rust/src/federation/queue/mod.rs

@@ -263,22 +263,30 @@ pub struct Completion {
     pub sent: Option<Sent>,
 }
 
-/// The parts of the loop that run in Python for now.
+/// Hears when the loop waits on the reactor: see [`LoopActivity`].
+pub type Observer = Arc<dyn AwaitableObserver>;
+
+/// The parts of the loop that run in Python for now. Each is given the
+/// calling loop's [`Observer`].
 #[async_trait]
 pub trait QueueHooks: Send + Sync + 'static {
     /// `_catch_up_transmission_loop`.
-    async fn catch_up(&self) -> Result<(), LoopError>;
+    async fn catch_up(&self, observer: &Observer) -> Result<(), LoopError>;
 
     /// The to-device messages and device list updates to send.
-    async fn device_edus(&self, request: DeviceEdusRequest) -> Result<DeviceEdus, LoopError>;
+    async fn device_edus(
+        &self,
+        observer: &Observer,
+        request: DeviceEdusRequest,
+    ) -> Result<DeviceEdus, LoopError>;
 
     /// `StickyEventBacklogTracker.prepare_transaction`: `None` once the
     /// backlog is empty.
-    async fn sticky_backlog(&self) -> Result<Option<StickyBatch>, LoopError>;
+    async fn sticky_backlog(&self, observer: &Observer) -> Result<Option<StickyBatch>, LoopError>;
 
     /// Record a finished transaction: the stream positions, the module
     /// callback and the metrics.
-    async fn complete(&self, completion: Completion) -> Result<(), LoopError>;
+    async fn complete(&self, observer: &Observer, completion: Completion) -> Result<(), LoopError>;
 }
 
 /// The answer to a sent transaction.
@@ -301,7 +309,7 @@ pub trait TransactionSink: Send + Sync + 'static {
     async fn send(
         &self,
         transaction: Transaction,
-        observer: Arc<dyn AwaitableObserver>,
+        observer: &Observer,
     ) -> Result<SendOutcome, LoopError>;
 }
 
@@ -319,8 +327,9 @@ impl Activity {
     }
 }
 
-/// One queue's part in [`Activity`]. Python calls tell it when they start
-/// and finish, on the reactor thread.
+/// One loop's part in [`Activity`]. Python calls tell it when they start
+/// and finish, on the reactor thread. Each loop has its own, so a loop that
+/// ends cannot clear the flag of the next one.
 pub struct LoopActivity {
     activity: Arc<Activity>,
     /// The loop is waiting on a Python call that has started.
@@ -428,7 +437,7 @@ pub struct DestinationQueue {
     state: Mutex<QueueState>,
     hooks: Arc<dyn QueueHooks>,
     sink: Arc<dyn TransactionSink>,
-    activity: Arc<LoopActivity>,
+    activity: Arc<Activity>,
     tokio: Handle,
 }
 
@@ -440,7 +449,7 @@ impl DestinationQueue {
         config: QueueConfig,
         hooks: Arc<dyn QueueHooks>,
         sink: Arc<dyn TransactionSink>,
-        activity: Arc<LoopActivity>,
+        activity: Arc<Activity>,
         tokio: Handle,
     ) -> Arc<Self> {
         if !config.should_send_on_this_instance {
@@ -654,7 +663,8 @@ impl DestinationQueue {
         let loop_id = NEXT_LOOP_ID.fetch_add(1, Ordering::Relaxed);
         state.transmission_loop_running = true;
         state.loop_id = loop_id;
-        self.activity.loop_started();
+        let activity = LoopActivity::new(Arc::clone(&self.activity));
+        activity.loop_started();
         // The guard takes the lock when it drops, which it does at once if
         // the runtime has shut down.
         drop(state);
@@ -663,11 +673,12 @@ impl DestinationQueue {
         let guard = LoopGuard {
             queue: Arc::clone(self),
             loop_id,
+            activity: Arc::clone(&activity),
         };
         let this = Arc::clone(self);
         let task = self.tokio.spawn(async move {
             let _guard = guard;
-            this.transmission_loop().await;
+            this.transmission_loop(activity).await;
         });
 
         let mut state = self.state();
@@ -677,15 +688,17 @@ impl DestinationQueue {
     }
 
     /// `_transaction_transmission_loop`.
-    async fn transmission_loop(&self) {
+    async fn transmission_loop(&self, activity: Arc<LoopActivity>) {
+        let observer: Observer = activity;
         let mut in_flight = Vec::new();
-        if let Err(err) = self.run_transmission_loop(&mut in_flight).await {
+        if let Err(err) = self.run_transmission_loop(&observer, &mut in_flight).await {
             self.handle_error(err, &in_flight);
         }
     }
 
     async fn run_transmission_loop(
         &self,
+        observer: &Observer,
         in_flight: &mut Vec<Arc<QueuedPdu>>,
     ) -> Result<(), LoopError> {
         // Fail fast if we are backing off. The client checks again.
@@ -694,7 +707,7 @@ impl DestinationQueue {
         }
 
         if self.state().catching_up {
-            self.hooks.catch_up().await?;
+            self.hooks.catch_up(observer).await?;
             if self.state().catching_up {
                 // Not caught up yet.
                 return Ok(());
@@ -711,7 +724,7 @@ impl DestinationQueue {
             }
             in_flight.clear();
 
-            let Some(prepared) = self.prepare_transaction().await? else {
+            let Some(prepared) = self.prepare_transaction(observer).await? else {
                 log::debug!(target: QUEUE_LOG, "TX [{}] Nothing to send", self.destination());
                 let mut state = self.state();
                 if state.new_data_to_send {
@@ -727,18 +740,23 @@ impl DestinationQueue {
             if !prepared.has_anything_to_send() {
                 // Filtering emptied a sticky event batch. Preparing it still
                 // moved the backlog on, which must be recorded.
-                self.complete_transaction(prepared, None).await?;
+                self.complete_transaction(observer, prepared, None).await?;
                 continue;
             }
 
             in_flight.extend(prepared.pdus.iter().cloned());
-            let sent = self.send_transaction(&prepared).await?;
-            self.complete_transaction(prepared, Some(sent)).await?;
+            let sent = self.send_transaction(observer, &prepared).await?;
+            self.complete_transaction(observer, prepared, Some(sent))
+                .await?;
         }
     }
 
     /// Build and send the transaction.
-    async fn send_transaction(&self, prepared: &Prepared) -> Result<Sent, LoopError> {
+    async fn send_transaction(
+        &self,
+        observer: &Observer,
+        prepared: &Prepared,
+    ) -> Result<Sent, LoopError> {
         let destination = self.destination();
         if !prepared.pdus.is_empty() {
             log::debug!(
@@ -757,7 +775,6 @@ impl DestinationQueue {
         let edus = prepared.edus.iter().cloned().map(Edu::into_value).collect();
         let transaction = self.sink.build(destination, pdus, edus);
 
-        let observer: Arc<dyn AwaitableObserver> = self.activity.clone();
         let outcome = self.sink.send(transaction, observer).await?;
         Ok(Sent {
             pdus: prepared.pdus.clone(),
@@ -827,16 +844,19 @@ impl DestinationQueue {
 
     /// `_prepare_transaction`: take turns between the sticky event backlog
     /// and the real-time queue, so neither starves the other.
-    async fn prepare_transaction(&self) -> Result<Option<Prepared>, LoopError> {
+    async fn prepare_transaction(
+        &self,
+        observer: &Observer,
+    ) -> Result<Option<Prepared>, LoopError> {
         let sticky_turn = std::mem::take(&mut self.state().sticky_backlog_turn);
         if sticky_turn {
-            if let Some(prepared) = self.prepare_sticky_transaction().await? {
+            if let Some(prepared) = self.prepare_sticky_transaction(observer).await? {
                 return Ok(Some(prepared));
             }
-            return self.prepare_main_queue_transaction().await;
+            return self.prepare_main_queue_transaction(observer).await;
         }
 
-        let prepared = self.prepare_main_queue_transaction().await?;
+        let prepared = self.prepare_main_queue_transaction(observer).await?;
         let backlogged = {
             let mut state = self.state();
             state.sticky_backlog_turn = self.config.msc4354_enabled && state.sticky_backlogged;
@@ -846,18 +866,21 @@ impl DestinationQueue {
             return Ok(prepared);
         }
         if backlogged {
-            return self.prepare_sticky_transaction().await;
+            return self.prepare_sticky_transaction(observer).await;
         }
         Ok(None)
     }
 
     /// `StickyEventBacklogTracker.prepare_transaction`.
-    async fn prepare_sticky_transaction(&self) -> Result<Option<Prepared>, LoopError> {
+    async fn prepare_sticky_transaction(
+        &self,
+        observer: &Observer,
+    ) -> Result<Option<Prepared>, LoopError> {
         if !self.config.msc4354_enabled || !self.state().sticky_backlogged {
             return Ok(None);
         }
 
-        let Some(batch) = self.hooks.sticky_backlog().await? else {
+        let Some(batch) = self.hooks.sticky_backlog(observer).await? else {
             log::info!(
                 target: QUEUE_LOG,
                 "Caught up on federation sticky event backlog for destination '{}'",
@@ -883,7 +906,10 @@ impl DestinationQueue {
     /// EDUs go in this order, up to 100: one presence EDU, up to five
     /// receipt EDUs, to-device messages (leaving ten places for device list
     /// updates), device list updates, then anything else.
-    async fn prepare_main_queue_transaction(&self) -> Result<Option<Prepared>, LoopError> {
+    async fn prepare_main_queue_transaction(
+        &self,
+        observer: &Observer,
+    ) -> Result<Option<Prepared>, LoopError> {
         let now_ms = self.sink.now_ms() as i64;
         let mut edus = Vec::new();
         let request = {
@@ -923,7 +949,7 @@ impl DestinationQueue {
             }
         };
 
-        let device = self.hooks.device_edus(request).await?;
+        let device = self.hooks.device_edus(observer, request).await?;
 
         let mut state = self.state();
         let mut edu_limit = request.edu_limit;
@@ -990,6 +1016,7 @@ impl DestinationQueue {
     /// and record how far we got.
     async fn complete_transaction(
         &self,
+        observer: &Observer,
         prepared: Prepared,
         sent: Option<Sent>,
     ) -> Result<(), LoopError> {
@@ -1012,13 +1039,16 @@ impl DestinationQueue {
         }
 
         self.hooks
-            .complete(Completion {
-                to_device_stream_id,
-                device_list_stream_id,
-                last_stream_ordering: prepared.last_stream_ordering.filter(|so| *so != 0),
-                sticky: prepared.sticky,
-                sent,
-            })
+            .complete(
+                observer,
+                Completion {
+                    to_device_stream_id,
+                    device_list_stream_id,
+                    last_stream_ordering: prepared.last_stream_ordering.filter(|so| *so != 0),
+                    sticky: prepared.sticky,
+                    sent,
+                },
+            )
             .await?;
 
         let mut state = self.state();
@@ -1094,6 +1124,7 @@ fn start_catching_up(state: &mut QueueState) {
 struct LoopGuard {
     queue: Arc<DestinationQueue>,
     loop_id: u64,
+    activity: Arc<LoopActivity>,
 }
 
 impl Drop for LoopGuard {
@@ -1105,6 +1136,6 @@ impl Drop for LoopGuard {
                 state.loop_task = None;
             }
         }
-        self.queue.activity.loop_ended();
+        self.activity.loop_ended();
     }
 }

rust/src/federation/queue/python.rs

@@ -20,7 +20,10 @@
 //! the [`QueueHooks`] as coroutines: `_rust_catch_up`, `_rust_device_edus`,
 //! `_rust_sticky_backlog` and `_rust_complete`.
 
-use std::sync::{Arc, Mutex};
+use std::{
+    sync::{Arc, Mutex},
+    time::Instant,
+};
 
 use async_trait::async_trait;
 use pyo3::{
@@ -31,16 +34,16 @@ use pyo3::{
 use serde_json::Value;
 
 use super::{
-    Activity, Completion, DestinationQueue, DeviceEdus, DeviceEdusRequest, Edu, LoopActivity,
-    LoopError, PduBody, PresenceState, QueueConfig, QueueHooks, QueuedPdu, Receipt, SendOutcome,
+    Activity, Completion, DestinationQueue, DeviceEdus, DeviceEdusRequest, Edu, LoopError,
+    Observer, PduBody, PresenceState, QueueConfig, QueueHooks, QueuedPdu, Receipt, SendOutcome,
     StickyBatch, StickyStreamPosition, TransactionSink,
 };
 use crate::{
-    deferred::{run_python_awaitable_observed, AwaitableObserver},
+    deferred::run_python_awaitable_observed,
     events::Event,
     federation::{
         client::{limiter::NotRetrying, RequestStats},
-        gil::{account, with_gil},
+        gil::{account, account_held, with_gil},
         transaction::{
             acknowledged_pdus,
             python::{python_acknowledged, RustTransaction, RustTransactionManager},
@@ -91,11 +94,9 @@ impl RustDestinationQueues {
             .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();
@@ -108,7 +109,7 @@ impl RustDestinationQueues {
                 },
                 Arc::new(hooks),
                 sink,
-                activity,
+                Arc::clone(&self.activity),
                 self.runtime.tokio_handle()?,
             ),
         })
@@ -311,15 +312,15 @@ 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.
+    /// with `convert` there. Both count as GIL time under `site`.
     async fn call<A, C, T>(
         &self,
+        observer: &Observer,
         site: &'static str,
         method: &'static str,
         args: A,
@@ -331,7 +332,6 @@ impl PyHooks {
         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| {
@@ -345,8 +345,13 @@ impl PyHooks {
                     facade.call_method1(method, args(py)?)
                 })
             },
-            convert,
-            Some(observer),
+            move |value| {
+                let start = Instant::now();
+                let converted = convert(value);
+                account_held(site, start.elapsed());
+                converted
+            },
+            Some(Arc::clone(observer)),
         )
         .await;
         result.map_err(|err| python_error(err, &self.destination))
@@ -355,8 +360,9 @@ impl PyHooks {
 
 #[async_trait]
 impl QueueHooks for PyHooks {
-    async fn catch_up(&self) -> Result<(), LoopError> {
+    async fn catch_up(&self, observer: &Observer) -> Result<(), LoopError> {
         self.call(
+            observer,
             "queue_catch_up",
             "_rust_catch_up",
             |py| Ok(PyTuple::empty(py)),
@@ -365,8 +371,13 @@ impl QueueHooks for PyHooks {
         .await
     }
 
-    async fn device_edus(&self, request: DeviceEdusRequest) -> Result<DeviceEdus, LoopError> {
+    async fn device_edus(
+        &self,
+        observer: &Observer,
+        request: DeviceEdusRequest,
+    ) -> Result<DeviceEdus, LoopError> {
         self.call(
+            observer,
             "queue_device_edus",
             "_rust_device_edus",
             move |py| {
@@ -409,8 +420,9 @@ impl QueueHooks for PyHooks {
         .await
     }
 
-    async fn sticky_backlog(&self) -> Result<Option<StickyBatch>, LoopError> {
+    async fn sticky_backlog(&self, observer: &Observer) -> Result<Option<StickyBatch>, LoopError> {
         self.call(
+            observer,
             "queue_sticky_backlog",
             "_rust_sticky_backlog",
             |py| Ok(PyTuple::empty(py)),
@@ -435,9 +447,10 @@ impl QueueHooks for PyHooks {
         .await
     }
 
-    async fn complete(&self, completion: Completion) -> Result<(), LoopError> {
+    async fn complete(&self, observer: &Observer, completion: Completion) -> Result<(), LoopError> {
         let completion = Arc::new(completion);
         self.call(
+            observer,
             "queue_complete",
             "_rust_complete",
             move |py| completion_args(py, &completion),
@@ -556,12 +569,24 @@ struct ClientSink {
 }
 
 impl ClientSink {
+    /// Record the request counters, which are Python's. A sent transaction's
+    /// go with its completion. A failed one needs this.
+    fn record_metrics(&self, py: Python<'_>, stats: &RequestStats) {
+        let recorded =
+            self.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}");
+        }
+    }
+
     /// Send through the Python stand-in.
     async fn send_for_test(
         &self,
         send: Arc<Py<PyAny>>,
         transaction: Transaction,
-        observer: Arc<dyn AwaitableObserver>,
+        observer: &Observer,
     ) -> Result<SendOutcome, LoopError> {
         let destination = transaction.destination.clone();
         let log_prefix = transaction.log_prefix();
@@ -581,7 +606,7 @@ impl ClientSink {
                     Err(_) => Value::Null,
                 })
             },
-            Some(observer),
+            Some(Arc::clone(observer)),
         )
         .await
         .map_err(|err| python_error(err, &destination))?;
@@ -618,7 +643,7 @@ impl TransactionSink for ClientSink {
     async fn send(
         &self,
         transaction: Transaction,
-        observer: Arc<dyn AwaitableObserver>,
+        observer: &Observer,
     ) -> Result<SendOutcome, LoopError> {
         let send_for_test = self.send_for_test.lock().expect("poisoned").clone();
         if let Some(send) = send_for_test {
@@ -636,6 +661,8 @@ impl TransactionSink for ClientSink {
             }),
             Ok(TransactionResponse::Unparsed(body)) => with_gil("queue_response_fallback", |py| {
                 python_acknowledged(py, &destination, &transaction_id, &body).map_err(|err| {
+                    // No later trip to Python will carry the counters.
+                    self.record_metrics(py, &stats);
                     classify_python_error(py, &err)
                         .unwrap_or_else(|_| LoopError::Other(format_exception(py, &err)))
                 })
@@ -645,19 +672,8 @@ impl TransactionSink for ClientSink {
                 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}");
-                    }
-                });
+                // As above, there is no later trip to Python.
+                with_gil("queue_send_failed", |py| self.record_metrics(py, &stats));
                 Err(LoopError::from_client(err, &destination))
             }
         }

rust/src/federation/queue/tests.rs

@@ -78,7 +78,7 @@ impl TransactionSink for FakeSink {
     async fn send(
         &self,
         transaction: Transaction,
-        _observer: Arc<dyn AwaitableObserver>,
+        _observer: &Observer,
     ) -> Result<SendOutcome, LoopError> {
         let answer = self
             .answers
@@ -133,7 +133,7 @@ impl FakeHooks {
 
 #[async_trait]
 impl QueueHooks for FakeHooks {
-    async fn catch_up(&self) -> Result<(), LoopError> {
+    async fn catch_up(&self, _observer: &Observer) -> Result<(), LoopError> {
         self.catch_ups.fetch_add(1, Ordering::SeqCst);
         // As Python's, with nothing in the database to catch up on.
         let queue = self.queue.get().and_then(Weak::upgrade).unwrap();
@@ -143,7 +143,11 @@ impl QueueHooks for FakeHooks {
         Ok(())
     }
 
-    async fn device_edus(&self, request: DeviceEdusRequest) -> Result<DeviceEdus, LoopError> {
+    async fn device_edus(
+        &self,
+        _observer: &Observer,
+        request: DeviceEdusRequest,
+    ) -> Result<DeviceEdus, LoopError> {
         self.device_requests.lock().unwrap().push(request);
         Ok(self
             .device_edus
@@ -157,11 +161,15 @@ impl QueueHooks for FakeHooks {
             }))
     }
 
-    async fn sticky_backlog(&self) -> Result<Option<StickyBatch>, LoopError> {
+    async fn sticky_backlog(&self, _observer: &Observer) -> Result<Option<StickyBatch>, LoopError> {
         Ok(self.sticky.lock().unwrap().pop_front().flatten())
     }
 
-    async fn complete(&self, completion: Completion) -> Result<(), LoopError> {
+    async fn complete(
+        &self,
+        _observer: &Observer,
+        completion: Completion,
+    ) -> Result<(), LoopError> {
         self.completed.lock().unwrap().push(Completed {
             to_device_stream_id: completion.to_device_stream_id,
             device_list_stream_id: completion.device_list_stream_id,
@@ -196,7 +204,7 @@ fn harness_with(msc4354_enabled: bool) -> Harness {
         },
         hooks.clone(),
         sink.clone(),
-        LoopActivity::new(Arc::clone(&activity)),
+        Arc::clone(&activity),
         Handle::current(),
     );
     hooks.queue.set(Arc::downgrade(&queue)).unwrap();
@@ -732,3 +740,21 @@ fn receipts_with_threads_share_an_edu() {
         .collect();
     assert_eq!(users, [["alice", "bob"], ["alice", "bob"]]);
 }
+
+#[test]
+fn a_loop_ending_does_not_touch_the_next() {
+    // The old loop ends after the next one has parked on Python.
+    let activity = Arc::new(Activity::default());
+    let old = LoopActivity::new(Arc::clone(&activity));
+    old.loop_started();
+    let next = LoopActivity::new(Arc::clone(&activity));
+    next.loop_started();
+    next.started();
+    old.loop_ended();
+    assert!(activity.idle());
+
+    next.finished();
+    assert!(!activity.idle());
+    next.loop_ended();
+    assert!(activity.idle());
+}

synapse/federation/sender/per_destination_queue.py

@@ -333,6 +333,14 @@ class _CatchUpLoop:
                     else:
                         room_catchup_pdus = [pdu]
 
+                # After a shutdown, the transaction manager returns without
+                # sending, and we would record the room as caught up. The
+                # Python loop is cancelled at shutdown, but the Rust one cannot
+                # cancel this coroutine. (mypy does not see that the awaits
+                # above can change the flag.)
+                if not self._transmission_loop_enabled:
+                    return  # type: ignore[unreachable]
+
                 logger.info(
                     "Catching up rooms to %s: %r", self._destination, pdu.room_id
                 )

tests/federation/test_federation_catch_up.py

@@ -375,6 +375,56 @@ class FederationCatchUpTestCases(_FederationCatchUpTestCaseBase):
             event_5.internal_metadata.stream_ordering,
         )
 
+    def test_catch_up_loop_stops_at_shutdown(self) -> None:
+        """A shutdown during catch-up stops it before the next room, so the
+        rooms it has not sent are not recorded as sent."""
+        per_dest_queue, sent_pdus = self.make_fake_destination_queue()
+        send = per_dest_queue._transaction_manager.send_new_transaction
+
+        async def send_then_shut_down(
+            destination: str, pdus: list[EventBase], edus: list[Edu]
+        ) -> None:
+            await send(destination, pdus, edus)
+            # What `shutdown` does to the loop. The transport client is a
+            # mock, which the transaction manager's shutdown would fail on.
+            per_dest_queue._transmission_loop_enabled = False
+
+        per_dest_queue._transaction_manager.send_new_transaction = send_then_shut_down  # type: ignore[method-assign]
+
+        self.register_user("u1", "you the one")
+        u1_token = self.login("u1", "you the one")
+        room_1 = self.helper.create_room_as("u1", tok=u1_token)
+        room_2 = self.helper.create_room_as("u1", tok=u1_token)
+        for room_id in (room_1, room_2):
+            self.get_success(
+                event_injection.inject_member_event(
+                    self.hs, room_id, "@user:host2", "join"
+                )
+            )
+
+        store = self.hs.get_datastores().main
+        event_id_0 = self.helper.send(room_1, "already sent", tok=u1_token)["event_id"]
+        event_id_1 = self.helper.send(room_1, "one", tok=u1_token)["event_id"]
+        self.helper.send(room_2, "two", tok=u1_token)
+        event_0 = self.get_success(store.get_event(event_id_0))
+        event_1 = self.get_success(store.get_event(event_id_1))
+        assert event_0.internal_metadata.stream_ordering is not None
+        self.get_success(
+            store.set_destination_last_successful_stream_ordering(
+                "host2", event_0.internal_metadata.stream_ordering
+            )
+        )
+
+        self.get_success(per_dest_queue._catch_up_transmission_loop())
+
+        self.assertEqual([pdu.event_id for pdu in sent_pdus], [event_id_1])
+        self.assertEqual(
+            self.get_success(
+                store.get_destination_last_successful_stream_ordering("host2")
+            ),
+            event_1.internal_metadata.stream_ordering,
+        )
+
     def test_catch_up_on_synapse_startup(self) -> None:
         """
         Tests the behaviour of get_catch_up_outstanding_destinations and

tests/federation/test_federation_sender.py

@@ -284,17 +284,17 @@ class FederationSenderReceiptsTestCases(HomeserverTestCase):
 
         self.pump()
 
-        # First, expect a call to send_transaction for the sending host
-        mock_send_transaction.assert_called()
-
-        transaction = mock_send_transaction.call_args_list[0].args[0]
-        self.assertEqual(transaction.destination, "host1")
-        self._assert_edu_in_call(mock_send_transaction.call_args_list[0].args[1])
-
-        # We also expect a call to one of the other hosts, as the first
-        # destination to wake up.
+        # Expect a call to send_transaction for the sending host, and one to
+        # one of the other hosts, as the first destination to wake up. With
+        # the queues in Rust, the two loops run at once, so either may send
+        # first.
         self.assertEqual(mock_send_transaction.call_count, 2)
-        self._assert_edu_in_call(mock_send_transaction.call_args.args[1])
+        destinations = [
+            call.args[0].destination for call in mock_send_transaction.call_args_list
+        ]
+        self.assertIn("host1", destinations)
+        for call in mock_send_transaction.call_args_list:
+            self._assert_edu_in_call(call.args[1])
 
         mock_send_transaction.reset_mock()