diff --git a/changelog.d/rust-fed-catchup.misc b/changelog.d/rust-fed-catchup.misc new file mode 100644 index 0000000000..9e295ad731 --- /dev/null +++ b/changelog.d/rust-fed-catchup.misc @@ -0,0 +1 @@ +Run federation catch-up and the sticky event backlog in Rust when the native Postgres driver is in use, under the experimental `rust_federation_sending: queues` option. diff --git a/rust/src/federation/queue/mod.rs b/rust/src/federation/queue/mod.rs index 05bd56108d..9b637fae4e 100644 --- a/rust/src/federation/queue/mod.rs +++ b/rust/src/federation/queue/mod.rs @@ -290,9 +290,13 @@ pub trait QueueHooks: Send + Sync + 'static { request: DeviceEdusRequest, ) -> Result; - /// `StickyEventBacklogTracker.prepare_transaction`: `None` once the - /// backlog is empty. - async fn sticky_backlog(&self, observer: &Observer) -> Result, LoopError>; + /// `StickyEventBacklogTracker.prepare_transaction`, for `queue`: `None` + /// once the backlog is empty. + async fn sticky_backlog( + &self, + queue: &DestinationQueue, + observer: &Observer, + ) -> Result, LoopError>; /// Record a finished transaction: the stream positions, the module /// callback and the metrics. @@ -908,7 +912,7 @@ impl DestinationQueue { return Ok(None); } - let Some(batch) = self.hooks.sticky_backlog(observer).await? else { + let Some(batch) = self.hooks.sticky_backlog(self, observer).await? else { log::info!( target: QUEUE_LOG, "Caught up on federation sticky event backlog for destination '{}'", @@ -1133,6 +1137,11 @@ impl DestinationQueue { self.state().sticky_backlogged = backlogged; } + /// The time now, by the sink's clock. + pub fn now_ms(&self) -> u64 { + self.sink.now_ms() + } + /// The event IDs of the queued PDUs, oldest first. For tests. pub fn pending_pdu_ids(&self) -> Vec { self.state() diff --git a/rust/src/federation/queue/native.rs b/rust/src/federation/queue/native.rs index e94f11dc36..15c49157f1 100644 --- a/rust/src/federation/queue/native.rs +++ b/rust/src/federation/queue/native.rs @@ -19,9 +19,8 @@ //! the device EDUs and records finished transactions with the //! [`FederationStore`], on tokio. A transaction then needs no trip to the //! reactor, unless it finishes a sticky event batch or a module wants to -//! hear about delivered events. Catch-up runs here too, reading events -//! through the shared event cache. The sticky event backlog stays Python, -//! through [`PyHooks`]. +//! hear about delivered events. Catch-up and the sticky event backlog run +//! here too, reading events through the shared event cache. use std::sync::{Arc, Mutex}; @@ -31,7 +30,7 @@ use pyo3::{exceptions::PyRuntimeError, intern, prelude::*}; use super::{ changes::StreamChanges, metrics::QueueMetrics, python::PyHooks, Completion, DestinationQueue, DeviceEdus, DeviceEdusRequest, LoopError, Observer, PduBody, QueueHooks, - QueuedPdu, StickyBatch, QUEUE_LOG, + QueuedPdu, StickyBatch, MAX_PDUS_PER_TRANSACTION, QUEUE_LOG, }; use crate::{ database::postgres::pool::{ConnectionPool, PyConnectionPool}, @@ -129,6 +128,9 @@ fn db_error(err: anyhow::Error) -> LoopError { LoopError::Other(format!("{err:#}")) } +/// A position in the sticky events stream, from the native store. +struct NativeStickyPosition(i64); + /// The stream ordering of an event read from the database, which has one. fn stream_ordering_of(event: &Event) -> Result { event @@ -428,8 +430,71 @@ impl QueueHooks for NativeHooks { }) } - async fn sticky_backlog(&self, observer: &Observer) -> Result, LoopError> { - self.python.sticky_backlog(observer).await + /// `StickyEventBacklogTracker.load_backlog`, natively. + async fn sticky_backlog( + &self, + queue: &DestinationQueue, + observer: &Observer, + ) -> Result, LoopError> { + let destination = self.destination.as_str(); + let pool = self.native.pool.get()?; + let Some((room_id, position, event_ids)) = self + .native + .store + .get_backlogged_sticky_events_for_destination( + &pool, + destination, + MAX_PDUS_PER_TRANSACTION, + queue.now_ms() as i64, + ) + .await + .map_err(db_error)? + else { + return Ok(None); + }; + log::debug!( + target: QUEUE_LOG, + "Selected {} backlogged sticky events to send to destination '{destination}' in room '{room_id}' (up to stream position {position})", + event_ids.len() + ); + + let reader = self.native.sync.reader(pool); + let ids: Vec> = event_ids.iter().map(|id| Arc::from(id.as_str())).collect(); + let events = reader + .events(&ids, RedactBehaviour::Redact, false, false) + .await + .map_err(db_error)?; + // Leave out what the destination may not see, rather than send it + // redacted, which would lose the stickiness. An erased user's + // sticky events are not sent in catch-up. They are all local + // events, so partial state does not apply. + let filter = ServerFilter { + target_server_name: destination, + local_server_name: &self.native.server_name, + redact: false, + filter_out_erased_senders: true, + filter_out_remote_partial_state_events: false, + }; + let events = match filter_events_for_server(&reader, &filter, events) + .await + .map_err(db_error)? + { + ServerFiltered::Events(events) => events, + // Python waits for the full state here. Let it. + ServerFiltered::NeedsFullState => { + return self.python.sticky_backlog(queue, observer).await; + } + }; + + let pdus = events + .into_iter() + .map(queued_pdu) + .collect::, _>>()?; + Ok(Some(StickyBatch { + room_id, + position: Arc::new(NativeStickyPosition(position)), + pdus, + })) } /// The database writes of `_rust_complete`, natively. Python is only @@ -494,9 +559,19 @@ impl QueueHooks for NativeHooks { } } - // The sticky batch finishes last, as in Python. - if completion.sticky.is_none() { + // The sticky batch finishes last, as in Python. One the native + // store loaded finishes natively, and one Python loaded in Python. + let Some((room_id, position)) = completion.sticky else { return Ok(()); + }; + if let Some(NativeStickyPosition(position)) = position.downcast_ref() { + let pool = self.native.pool.get()?; + return self + .native + .store + .mark_backlogged_sticky_events_sent(&pool, destination, &room_id, *position) + .await + .map_err(db_error); } self.python .call_complete( @@ -505,7 +580,7 @@ impl QueueHooks for NativeHooks { to_device_stream_id: None, device_list_stream_id: None, last_stream_ordering: None, - sticky: completion.sticky, + sticky: Some((room_id, position)), sent: None, }, ) diff --git a/rust/src/federation/queue/python.rs b/rust/src/federation/queue/python.rs index a759e493c6..85a409e99f 100644 --- a/rust/src/federation/queue/python.rs +++ b/rust/src/federation/queue/python.rs @@ -550,7 +550,11 @@ impl QueueHooks for PyHooks { .await } - async fn sticky_backlog(&self, observer: &Observer) -> Result, LoopError> { + async fn sticky_backlog( + &self, + _queue: &DestinationQueue, + observer: &Observer, + ) -> Result, LoopError> { self.call( observer, "queue_sticky_backlog", diff --git a/rust/src/federation/queue/tests.rs b/rust/src/federation/queue/tests.rs index 3452a6f6b6..7f1544ef30 100644 --- a/rust/src/federation/queue/tests.rs +++ b/rust/src/federation/queue/tests.rs @@ -165,7 +165,11 @@ impl QueueHooks for FakeHooks { })) } - async fn sticky_backlog(&self, _observer: &Observer) -> Result, LoopError> { + async fn sticky_backlog( + &self, + _queue: &DestinationQueue, + _observer: &Observer, + ) -> Result, LoopError> { Ok(self.sticky.lock().unwrap().pop_front().flatten()) } diff --git a/rust/src/storage/federation_store.rs b/rust/src/storage/federation_store.rs index ebb07bc0ea..2f77defcec 100644 --- a/rust/src/storage/federation_store.rs +++ b/rust/src/storage/federation_store.rs @@ -506,6 +506,199 @@ impl FederationStore { .collect() } + /// `get_backlogged_sticky_events_for_destination`: up to `limit` of + /// the local sticky events, live at `now_ms`, from one room whose + /// backlog for `destination` is not empty, oldest first, and the sticky + /// events stream position of the last. `None` if there is no backlog. + /// + /// A room in the backlog whose sticky events have all expired is + /// removed from it, and another room tried. + pub async fn get_backlogged_sticky_events_for_destination( + &self, + pool: &ConnectionPool, + destination: &str, + limit: usize, + now_ms: i64, + ) -> anyhow::Result)>> { + let pattern = self.config.local_users_pattern(); + let mut conn = Conn::borrow(pool).await?; + conn.begin().await?; + + let mut backlog = self + .try_backlogged_sticky_events(&conn, destination, limit, now_ms, &pattern) + .await?; + if matches!(&backlog, Some((_, _, events)) if events.is_empty()) { + // The room is backlogged but has no sticky events to send, as + // they expired, say. Clean up and try again. + conn.query( + "WITH to_clean_up AS ( + SELECT backlog.room_id + FROM destination_rooms_sticky_events_backlog AS backlog + LEFT JOIN sticky_events AS se + ON se.room_id = backlog.room_id + AND se.sender LIKE $1 + AND $2 < se.expires_at + AND backlog.sticky_events_stream_position <= se.stream_id + WHERE se.event_id IS NULL + AND backlog.destination = $3 + ) + DELETE FROM destination_rooms_sticky_events_backlog + WHERE destination = $3 AND room_id IN (SELECT room_id FROM to_clean_up)", + &[&pattern, &now_ms, &destination], + ) + .await?; + backlog = self + .try_backlogged_sticky_events(&conn, destination, limit, now_ms, &pattern) + .await?; + if matches!(&backlog, Some((_, _, events)) if events.is_empty()) { + anyhow::bail!("Sticky event backlog for {destination} not cleaned up"); + } + } + conn.commit().await?; + + Ok(backlog.map(|(room_id, position, events)| { + (room_id, position.unwrap_or_default(), events) + })) + } + + /// One try of [`FederationStore::get_backlogged_sticky_events_for_destination`]: + /// a backlogged room, the position of its last event and its events, + /// which may be none. + async fn try_backlogged_sticky_events( + &self, + conn: &Conn, + destination: &str, + limit: usize, + now_ms: i64, + pattern: &str, + ) -> anyhow::Result, Vec)>> { + let rows = conn + .query( + "SELECT room_id, sticky_events_stream_position::bigint + FROM destination_rooms_sticky_events_backlog + WHERE destination = $1 + LIMIT 1", + &[&destination], + ) + .await?; + let Some(row) = rows.first() else { + return Ok(None); + }; + let room_id: String = row.try_get(0)?; + let from_position: i64 = row.try_get(1)?; + + let rows = conn + .query( + "SELECT event_id, stream_id::bigint + FROM sticky_events + WHERE room_id = $1 + AND $2::bigint <= stream_id + AND $3 < expires_at + AND sender LIKE $4 + ORDER BY stream_id ASC + LIMIT $5", + &[&room_id, &from_position, &now_ms, &pattern, &(limit as i64)], + ) + .await?; + let mut position = None; + let mut events = Vec::with_capacity(rows.len()); + for row in &rows { + events.push(row.try_get(0)?); + let stream_id: i64 = row.try_get(1)?; + position = Some(position.map_or(stream_id, |p: i64| p.max(stream_id))); + } + Ok(Some((room_id, position, events))) + } + + /// `mark_backlogged_sticky_events_sent`: the sticky events of the room + /// up to `position` reached `destination`. + pub async fn mark_backlogged_sticky_events_sent( + &self, + pool: &ConnectionPool, + destination: &str, + room_id: &str, + position: i64, + ) -> anyhow::Result<()> { + Conn::borrow(pool) + .await? + .query( + "INSERT INTO destination_rooms_sticky_events_backlog + (destination, room_id, sticky_events_stream_position) + VALUES ($1, $2, $3::bigint) + ON CONFLICT (destination, room_id) DO UPDATE + SET sticky_events_stream_position = EXCLUDED.sticky_events_stream_position", + // The position is an inclusive lower bound on what is left. + &[&destination, &room_id, &(position + 1)], + ) + .await?; + Ok(()) + } + + /// `get_catch_up_outstanding_destinations`: up to 25 destinations after + /// `after_destination`, in order, that we are not backing off from at + /// `now_ms` and that have PDUs, to-device messages or sticky events + /// waiting. + pub async fn get_catch_up_outstanding_destinations( + &self, + pool: &ConnectionPool, + after_destination: Option<&str>, + now_ms: i64, + ) -> anyhow::Result> { + // Everything sorts after "". + let after = after_destination.unwrap_or(""); + let rows = Conn::borrow(pool) + .await? + .query( + "WITH pdu_destinations AS ( + SELECT DISTINCT destination FROM destination_rooms + LEFT JOIN destinations USING (destination) + WHERE + destination > $1 + AND destination_rooms.stream_ordering + > COALESCE(destinations.last_successful_stream_ordering, 0) + AND ( + destinations.retry_last_ts IS NULL OR + destinations.retry_last_ts + destinations.retry_interval < $2 + ) + ORDER BY destination + LIMIT 25 + ), to_device_destinations AS ( + SELECT DISTINCT destination FROM device_federation_outbox + LEFT JOIN destinations USING (destination) + WHERE + destination > $1 + AND ( + destinations.retry_last_ts IS NULL OR + destinations.retry_last_ts + destinations.retry_interval < $2 + ) + ORDER BY destination + LIMIT 25 + ), sticky_event_destinations AS ( + SELECT DISTINCT destination + FROM destination_rooms_sticky_events_backlog + LEFT JOIN destinations USING (destination) + WHERE + destination > $1 + AND ( + destinations.retry_last_ts IS NULL OR + destinations.retry_last_ts + destinations.retry_interval < $2 + ) + ORDER BY destination + LIMIT 25 + ) + SELECT destination FROM pdu_destinations + UNION SELECT destination FROM to_device_destinations + UNION SELECT destination FROM sticky_event_destinations + ORDER BY destination + LIMIT 25", + &[&after, &now_ms], + ) + .await?; + rows.iter() + .map(|row| Ok(row.try_get(0)?)) + .collect() + } + /// `mark_backlogged_sticky_events_after_catchup_transaction`: a /// catch-up transaction moved `destination` on from `old_position` to /// `new_position`, skipping events. Record each room with local sticky @@ -1055,6 +1248,53 @@ impl PyFederationStoreForTest { self.run(py, self.store.get_prev_events_for_room(&self.pool, room_id)) } + #[allow(clippy::type_complexity)] + fn get_backlogged_sticky_events_for_destination( + &self, + py: Python<'_>, + destination: &str, + limit: usize, + now_ms: i64, + ) -> PyResult)>> { + self.run( + py, + self.store.get_backlogged_sticky_events_for_destination( + &self.pool, + destination, + limit, + now_ms, + ), + ) + } + + fn mark_backlogged_sticky_events_sent( + &self, + py: Python<'_>, + destination: &str, + room_id: &str, + position: i64, + ) -> PyResult<()> { + self.run( + py, + self.store + .mark_backlogged_sticky_events_sent(&self.pool, destination, room_id, position), + ) + } + + #[pyo3(signature = (after_destination, now_ms))] + fn get_catch_up_outstanding_destinations( + &self, + py: Python<'_>, + after_destination: Option<&str>, + now_ms: i64, + ) -> PyResult> { + self.run( + py, + self.store + .get_catch_up_outstanding_destinations(&self.pool, after_destination, now_ms), + ) + } + fn mark_backlogged_sticky_events_after_catchup_transaction( &self, py: Python<'_>, diff --git a/synapse/synapse_rust/federation.pyi b/synapse/synapse_rust/federation.pyi index d6e4f0e4aa..fd1df1e1f0 100644 --- a/synapse/synapse_rust/federation.pyi +++ b/synapse/synapse_rust/federation.pyi @@ -323,6 +323,15 @@ class FederationStoreForTest: self, destination: str, last_successful_stream_ordering: int ) -> list[str]: ... def get_prev_events_for_room(self, room_id: str) -> list[str]: ... + def get_backlogged_sticky_events_for_destination( + self, destination: str, limit: int, now_ms: int + ) -> tuple[str, int, list[str]] | None: ... + def mark_backlogged_sticky_events_sent( + self, destination: str, room_id: str, position: int + ) -> None: ... + def get_catch_up_outstanding_destinations( + self, after_destination: str | None, now_ms: int + ) -> list[str]: ... def mark_backlogged_sticky_events_after_catchup_transaction( self, destination: str, diff --git a/tests/federation/test_rust_queues.py b/tests/federation/test_rust_queues.py index d6156a8c39..b100cad7aa 100644 --- a/tests/federation/test_rust_queues.py +++ b/tests/federation/test_rust_queues.py @@ -51,6 +51,7 @@ from synapse.synapse_rust.federation import ( federation_gil_stats, ) from synapse.util.clock import Clock +from synapse.util.duration import Duration from tests import unittest from tests.federation._rust_transactions import wait_for_rust_queues @@ -118,7 +119,9 @@ class RustQueuesTestCase(unittest.HomeserverTestCase): def _make_queues( self, transactions: RustTransactionManager ) -> RustDestinationQueues: - return RustDestinationQueues(transactions, False, {DEST}) + return RustDestinationQueues( + transactions, self.hs.config.experimental.msc4354_enabled, {DEST} + ) def _send_event(self, body: str) -> EventBase: event_id = self.helper.send(self.room_id, body, tok=self.tok)["event_id"] @@ -264,7 +267,13 @@ class RustQueuesTestCase(unittest.HomeserverTestCase): class RustQueuesNativeTestCase(RustQueuesTestCase): - """The same, with the database work done through the native driver.""" + """The same, with the database work done through the native driver, and + sticky events on, so the loops look at the sticky event backlog.""" + + def default_config(self) -> dict[str, Any]: + config = super().default_config() + config["experimental_features"]["msc4354_enabled"] = True + return config def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer: hs = self.setup_test_homeserver() @@ -394,6 +403,57 @@ class RustQueuesNativeTestCase(RustQueuesTestCase): [self.event.content["body"]], ) + def test_sticky_backlog(self) -> None: + """Catching up past a sticky event backlogs it, and the loop then + sends it, all without taking the GIL.""" + self.hs.get_module_api_callbacks().federation._on_event_delivered_over_federation_callbacks.clear() + destination = "sticky.example.com" + self._catch_up_destination(destination, self.event) + self.helper.send_sticky_event( + self.room_id, + "m.room.message", + duration=Duration(minutes=10), + content={"body": "sticky", "msgtype": "m.text"}, + tok=self.tok, + ) + latest = self._send_event("latest") + stream_ordering = latest.internal_metadata.stream_ordering + assert stream_ordering is not None + self.get_success( + self.store.store_destination_rooms_entries( + [destination], self.room_id, stream_ordering + ) + ) + queue = self.sender._get_per_destination_queue(destination) + assert isinstance(queue, RustPerDestinationQueue) + self.transport.responses.extend([json_response(200, {})] * 2) + + before = federation_gil_stats() + queue.attempt_new_transaction() + self._settle() + + bodies = [ + [pdu["content"].get("body") for pdu in json.loads(request.body)["pdus"]] + for request in self.transport.requests + ] + self.assertEqual(bodies, [["latest"], ["sticky"]]) + # The loop looked again, found the backlog sent, and cleared it. + self.assertEqual( + self.get_success( + self.store.db_pool.simple_select_list( + "destination_rooms_sticky_events_backlog", + {"destination": destination}, + ["room_id"], + ) + ), + [], + ) + self.assertFalse(queue._rust.sticky_backlogged) + calls = self._queue_calls(before) + self.assertNotIn("queue_catch_up", calls) + self.assertNotIn("queue_sticky_backlog", calls) + self.assertNotIn("queue_complete", calls) + def test_to_device(self) -> None: """A to-device message is sent, and deleted once sent. Reading the stream position is the one GIL crossing.""" diff --git a/tests/storage/test_federation_store.py b/tests/storage/test_federation_store.py index 002316958f..c212560534 100644 --- a/tests/storage/test_federation_store.py +++ b/tests/storage/test_federation_store.py @@ -31,12 +31,13 @@ from signedjson.key import ( from twisted.internet.testing import MemoryReactor from synapse.handlers.e2e_keys import SignatureListItem +from synapse.replication.tcp.streams._base import StickyEventStreamPosition from synapse.rest import admin from synapse.rest.client import login, room from synapse.server import HomeServer from synapse.storage.rust_pool import RustConnectionPool from synapse.synapse_rust.federation import FederationStoreForTest -from synapse.types import JsonDict +from synapse.types import JsonDict, RoomID from synapse.util.clock import Clock from synapse.util.duration import Duration @@ -540,3 +541,132 @@ class CatchUpStoreTestCase(unittest.HomeserverTestCase): self.assertEqual(self._backlog(), python, (old, new, sent, existing)) # The cases did backlog something. self.assertNotEqual(python, []) + + def _seed_backlog(self) -> list[int]: + """Sticky and plain events in each room, with every room backlogged + for `DEST` from its start.""" + orderings = [self._send(self.rooms[i % 3], sticky=i % 2 == 0) for i in range(9)] + self.get_success( + self.store.mark_backlogged_sticky_events_after_catchup_transaction( + DEST, + old_last_successfully_sent_stream_ordering=0, + new_last_successfully_sent_stream_ordering=orderings[-1] + 1, + event_stream_orderings_sent_in_transaction=[], + ) + ) + return orderings + + def test_backlogged_sticky_events_for_destination(self) -> None: + self._seed_backlog() + now = self.clock.time_msec() + for limit in (1, 2, 50): + python = self.get_success( + self.store.get_backlogged_sticky_events_for_destination( + DEST, limit=limit + ) + ) + assert python is not None + room_id, position, event_ids = python + self.assertEqual( + self.rust.get_backlogged_sticky_events_for_destination( + DEST, limit, now + ), + (room_id.to_string(), position, event_ids), + ) + self.assertIsNone( + self.rust.get_backlogged_sticky_events_for_destination(OTHER, 50, now) + ) + + def test_backlogged_sticky_events_expire(self) -> None: + """Once the sticky events expire, the backlog is cleaned up.""" + self._seed_backlog() + self.assertNotEqual(self._backlog(), []) + later = self.clock.time_msec() + 60 * 60 * 1000 + self.assertIsNone( + self.rust.get_backlogged_sticky_events_for_destination(DEST, 50, later) + ) + self.assertEqual(self._backlog(), []) + + def test_mark_backlogged_sticky_events_sent(self) -> None: + self._seed_backlog() + backlog = self.rust.get_backlogged_sticky_events_for_destination( + DEST, 50, self.clock.time_msec() + ) + assert backlog is not None + room_id, position, _ = backlog + + self.rust.mark_backlogged_sticky_events_sent(DEST, room_id, position) + rust = self._backlog() + self._clear_backlog() + self._seed_backlog_rows(rust) + self.get_success( + self.store.mark_backlogged_sticky_events_sent( + DEST, RoomID.from_string(room_id), StickyEventStreamPosition(position) + ) + ) + self.assertEqual(self._backlog(), rust) + self.assertIn((DEST, room_id, position + 1), rust) + + def _seed_backlog_rows(self, rows: list[tuple[str, str, int]]) -> None: + for destination, room_id, position in rows: + self.get_success( + self.store.db_pool.simple_upsert( + "destination_rooms_sticky_events_backlog", + {"destination": destination, "room_id": room_id}, + {"sticky_events_stream_position": position}, + ) + ) + + def test_catch_up_outstanding_destinations(self) -> None: + # PDUs for `DEST` and `OTHER`, a to-device message for a third, a + # sticky backlog for a fourth, and a fifth that is backed off. + self._send(self.rooms[0], sticky=False) + self.get_success( + self.store.add_remote_messages_from_client_to_device_inbox( + { + "todevice.example.com": { + "sender": "@alice:test", + "type": "m.test", + "message_id": "1", + "messages": {"@u:todevice.example.com": {"D": {}}}, + } + } + ) + ) + self._seed_backlog_rows([("sticky.example.com", self.rooms[0], 1)]) + self.get_success( + self.store.store_destination_rooms_entries( + ["backoff.example.com"], self.rooms[0], 1_000_000 + ) + ) + now = self.clock.time_msec() + self.get_success( + self.store.set_destination_retry_timings( + "backoff.example.com", now, now, 60_000 + ) + ) + + for after in (None, "", DEST, "zzz"): + for at in (now, now + 120_000): + # Python's reads the clock, so call its transaction directly. + python = self.get_success( + self.store.db_pool.runInteraction( + "outstanding", + self.store._get_catch_up_outstanding_destinations_txn, + at, + after, + ) + ) + self.assertEqual( + self.rust.get_catch_up_outstanding_destinations(after, at), + python, + (after, at), + ) + self.assertIn( + "backoff.example.com", + self.rust.get_catch_up_outstanding_destinations(None, now + 120_000), + ) + self.assertNotIn( + "backoff.example.com", + self.rust.get_catch_up_outstanding_destinations(None, now), + )