From 2d76b50a9c15c8bd3cd8fc2efe7655e411f8696f Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Thu, 15 Oct 2026 17:03:57 +0000 Subject: [PATCH] Send the sticky event backlog in Rust on the native driver `NativeHooks::sticky_backlog` loads a backlogged batch, cleaning up rooms whose sticky events expired, reads the events through the shared event cache and filters them for the destination. Its completion records the batch natively. Where the filter would wait for full state, the batch is left to Python, as before. `FederationStore` also gains `get_catch_up_outstanding_destinations` for phase 6. Each query has a differential test against Python's. The native queue tests now run with MSC4354 on, so the loops look at the sticky event backlog, and pin that it takes no GIL. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01R5b7Ruy4cgt27Mjq6Ptxbb --- changelog.d/rust-fed-catchup.misc | 1 + rust/src/federation/queue/mod.rs | 17 +- rust/src/federation/queue/native.rs | 93 +++++++++- rust/src/federation/queue/python.rs | 6 +- rust/src/federation/queue/tests.rs | 6 +- rust/src/storage/federation_store.rs | 240 +++++++++++++++++++++++++ synapse/synapse_rust/federation.pyi | 9 + tests/federation/test_rust_queues.py | 64 ++++++- tests/storage/test_federation_store.py | 132 +++++++++++++- 9 files changed, 550 insertions(+), 18 deletions(-) create mode 100644 changelog.d/rust-fed-catchup.misc 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), + )