diff --git a/rust/src/deferred.rs b/rust/src/deferred.rs index f0fa396d0a..8fed063ec4 100644 --- a/rust/src/deferred.rs +++ b/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); } diff --git a/rust/src/federation/gil.rs b/rust/src/federation/gil.rs index 8c5534d555..87d7aa4216 100644 --- a/rust/src/federation/gil.rs +++ b/rust/src/federation/gil.rs @@ -62,13 +62,24 @@ pub fn account(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)); + } } diff --git a/rust/src/federation/queue/mod.rs b/rust/src/federation/queue/mod.rs index ab60116f0b..a76bd377da 100644 --- a/rust/src/federation/queue/mod.rs +++ b/rust/src/federation/queue/mod.rs @@ -263,22 +263,30 @@ pub struct Completion { pub sent: Option, } -/// 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; + +/// 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; + async fn device_edus( + &self, + observer: &Observer, + request: DeviceEdusRequest, + ) -> Result; /// `StickyEventBacklogTracker.prepare_transaction`: `None` once the /// backlog is empty. - async fn sticky_backlog(&self) -> Result, LoopError>; + async fn sticky_backlog(&self, observer: &Observer) -> Result, 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, + observer: &Observer, ) -> Result; } @@ -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, /// The loop is waiting on a Python call that has started. @@ -428,7 +437,7 @@ pub struct DestinationQueue { state: Mutex, hooks: Arc, sink: Arc, - activity: Arc, + activity: Arc, tokio: Handle, } @@ -440,7 +449,7 @@ impl DestinationQueue { config: QueueConfig, hooks: Arc, sink: Arc, - activity: Arc, + activity: Arc, tokio: Handle, ) -> Arc { 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) { + 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>, ) -> 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 { + async fn send_transaction( + &self, + observer: &Observer, + prepared: &Prepared, + ) -> Result { 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 = 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, LoopError> { + async fn prepare_transaction( + &self, + observer: &Observer, + ) -> Result, 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, LoopError> { + async fn prepare_sticky_transaction( + &self, + observer: &Observer, + ) -> Result, 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, LoopError> { + async fn prepare_main_queue_transaction( + &self, + observer: &Observer, + ) -> Result, 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, ) -> 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, loop_id: u64, + activity: Arc, } 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(); } } diff --git a/rust/src/federation/queue/python.rs b/rust/src/federation/queue/python.rs index 4a1b3b122a..c0408a0163 100644 --- a/rust/src/federation/queue/python.rs +++ b/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 = 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>, - activity: Arc, destination: String, } impl PyHooks { /// Await `facade.(*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( &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 = 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 { + async fn device_edus( + &self, + observer: &Observer, + request: DeviceEdusRequest, + ) -> Result { 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, LoopError> { + async fn sticky_backlog(&self, observer: &Observer) -> Result, 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>, transaction: Transaction, - observer: Arc, + observer: &Observer, ) -> Result { 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, + observer: &Observer, ) -> Result { 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)) } } diff --git a/rust/src/federation/queue/tests.rs b/rust/src/federation/queue/tests.rs index 7f05fd2994..56f0e3510d 100644 --- a/rust/src/federation/queue/tests.rs +++ b/rust/src/federation/queue/tests.rs @@ -78,7 +78,7 @@ impl TransactionSink for FakeSink { async fn send( &self, transaction: Transaction, - _observer: Arc, + _observer: &Observer, ) -> Result { 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 { + async fn device_edus( + &self, + _observer: &Observer, + request: DeviceEdusRequest, + ) -> Result { 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, LoopError> { + async fn sticky_backlog(&self, _observer: &Observer) -> Result, 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()); +} diff --git a/synapse/federation/sender/per_destination_queue.py b/synapse/federation/sender/per_destination_queue.py index 8bd7cf9ed3..f70d54d45e 100644 --- a/synapse/federation/sender/per_destination_queue.py +++ b/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 ) diff --git a/tests/federation/test_federation_catch_up.py b/tests/federation/test_federation_catch_up.py index 10415c6fa3..b0e5d34a99 100644 --- a/tests/federation/test_federation_catch_up.py +++ b/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 diff --git a/tests/federation/test_federation_sender.py b/tests/federation/test_federation_sender.py index 160a0f4c58..89316b29ca 100644 --- a/tests/federation/test_federation_sender.py +++ b/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()