diff --git a/rust/src/federation/queue/mod.rs b/rust/src/federation/queue/mod.rs index c57688f9c1..05bd56108d 100644 --- a/rust/src/federation/queue/mod.rs +++ b/rust/src/federation/queue/mod.rs @@ -156,6 +156,8 @@ impl Receipt { pub enum PduBody { /// The event. Its JSON is read on tokio, without the GIL. Event(Py), + /// An event loaded in Rust, for catch-up. + Native(Event), /// The JSON itself, for tests. #[cfg(test)] Json(Value), @@ -174,6 +176,7 @@ impl QueuedPdu { fn to_value(&self) -> serde_json::Result { match &self.body { PduBody::Event(event) => event.get().to_pdu_value(), + PduBody::Native(event) => event.to_pdu_value(), #[cfg(test)] PduBody::Json(value) => Ok(value.clone()), } @@ -276,8 +279,9 @@ pub type Observer = Arc; /// calling loop's [`Observer`]. #[async_trait] pub trait QueueHooks: Send + Sync + 'static { - /// `_catch_up_transmission_loop`. - async fn catch_up(&self, observer: &Observer) -> Result<(), LoopError>; + /// `_catch_up_transmission_loop`, for `queue`. + async fn catch_up(&self, queue: &DestinationQueue, observer: &Observer) + -> Result<(), LoopError>; /// The to-device messages and device list updates to send. async fn device_edus( @@ -713,7 +717,7 @@ impl DestinationQueue { } if self.state().catching_up { - self.hooks.catch_up(observer).await?; + self.hooks.catch_up(self, observer).await?; if self.state().catching_up { // Not caught up yet. return Ok(()); @@ -794,6 +798,24 @@ impl DestinationQueue { }) } + /// Send `pdus` in a transaction of their own, as catch-up does. The + /// caller records it. + pub(super) async fn send_pdus( + &self, + observer: &Observer, + pdus: Vec>, + ) -> Result { + let prepared = Prepared { + pdus, + edus: Vec::new(), + to_device_stream_id: None, + device_list_stream_id: None, + last_stream_ordering: None, + sticky: None, + }; + self.send_transaction(observer, &prepared).await + } + /// The exception handlers of `_transaction_transmission_loop`. fn handle_error(&self, err: LoopError, in_flight: &[Arc]) { let destination = self.destination(); diff --git a/rust/src/federation/queue/native.rs b/rust/src/federation/queue/native.rs index 62995540c7..e94f11dc36 100644 --- a/rust/src/federation/queue/native.rs +++ b/rust/src/federation/queue/native.rs @@ -19,8 +19,9 @@ //! 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 and the sticky event backlog stay -//! Python, through [`PyHooks`]. +//! hear about delivered events. Catch-up runs here too, reading events +//! through the shared event cache. The sticky event backlog stays Python, +//! through [`PyHooks`]. use std::sync::{Arc, Mutex}; @@ -28,13 +29,20 @@ use async_trait::async_trait; use pyo3::{exceptions::PyRuntimeError, intern, prelude::*}; use super::{ - changes::StreamChanges, metrics::QueueMetrics, python::PyHooks, Completion, DeviceEdus, - DeviceEdusRequest, LoopError, Observer, QueueHooks, StickyBatch, + changes::StreamChanges, metrics::QueueMetrics, python::PyHooks, Completion, + DestinationQueue, DeviceEdus, DeviceEdusRequest, LoopError, Observer, PduBody, QueueHooks, + QueuedPdu, StickyBatch, QUEUE_LOG, }; use crate::{ database::postgres::pool::{ConnectionPool, PyConnectionPool}, + events::Event, federation::gil::with_gil, - storage::federation_store::FederationStore, + storage::{ + events::RedactBehaviour, + federation_store::FederationStore, + sync_store::{SyncReader, SyncStore}, + }, + visibility::{filter_events_for_server, ServerFilter, ServerFiltered}, }; /// The logger of `synapse.logging.issue9533_logger`. @@ -51,6 +59,11 @@ pub struct Native { pub to_device: StreamChanges, /// Which destinations have new device list updates. pub device_lists: StreamChanges, + /// The queries over the shared event and state caches, which the sync + /// handler uses too. + pub sync: Arc, + /// This server's name. + pub server_name: String, } impl Native { @@ -116,10 +129,218 @@ fn db_error(err: anyhow::Error) -> LoopError { LoopError::Other(format!("{err:#}")) } +/// The stream ordering of an event read from the database, which has one. +fn stream_ordering_of(event: &Event) -> Result { + event + .internal_metadata() + .stream_ordering() + .map(|so| so.get()) + .ok_or_else(|| LoopError::Other(format!("Event {} has no stream ordering", event.event_id()))) +} + +fn queued_pdu(event: Event) -> Result, LoopError> { + Ok(Arc::new(QueuedPdu { + event_id: event.event_id().to_owned(), + stream_ordering: Some(stream_ordering_of(&event)?), + origin_server_ts: event.origin_server_ts(), + body: PduBody::Native(event), + })) +} + +impl NativeHooks { + /// The PDUs to catch `pdu`'s room up with: the room's latest forward + /// extremities that the destination may see, or `pdu` itself. + async fn room_catch_up_pdus( + &self, + reader: &SyncReader<'_>, + pool: &ConnectionPool, + pdu: Event, + last_successful_stream_ordering: i64, + ) -> Result, LoopError> { + // The PDU from the database is the newest PDU in the room from this + // server that we failed to send. Other servers may have sent lots + // since, and we would rather tell the destination about the latest + // events in the room only. + let extremities = self + .native + .store + .get_prev_events_for_room(pool, pdu.room_id()) + .await + .map_err(db_error)?; + if extremities.iter().any(|id| id == pdu.event_id()) { + return Ok(vec![pdu]); + } + if reader + .is_partial_state_room(pdu.room_id()) + .await + .map_err(db_error)? + { + // We cannot be sure which events the destination may see with + // partial state, so retry the newest PDU it is missing from us. + return Ok(vec![pdu]); + } + + let ids: Vec> = extremities.iter().map(|id| Arc::from(id.as_str())).collect(); + let extremities = reader + .events(&ids, RedactBehaviour::Redact, false, false) + .await + .map_err(db_error)?; + let mut new_pdus = Vec::with_capacity(extremities.len()); + for event in extremities { + // Leave out events from before the destination went offline. + if stream_ordering_of(&event)? >= last_successful_stream_ordering { + new_pdus.push(event); + } + } + + // Leave out events the destination may not see, such as when it has + // left the room. + let filter = ServerFilter { + target_server_name: &self.destination, + local_server_name: &self.native.server_name, + redact: false, + filter_out_erased_senders: true, + filter_out_remote_partial_state_events: true, + }; + match filter_events_for_server(reader, &filter, new_pdus) + .await + .map_err(db_error)? + { + ServerFiltered::Events(events) if !events.is_empty() => Ok(events), + // With every extremity filtered out, send the original event, so + // the destination gets at least some of what it missed. Python + // would wait for the full state where Rust gives up, and we do + // the same as for a room with partial state then. + ServerFiltered::Events(_) | ServerFiltered::NeedsFullState => Ok(vec![pdu]), + } + } +} + #[async_trait] impl QueueHooks for NativeHooks { - async fn catch_up(&self, observer: &Observer) -> Result<(), LoopError> { - self.python.catch_up(observer).await + /// `_catch_up_transmission_loop`, natively. + async fn catch_up(&self, queue: &DestinationQueue, observer: &Observer) -> Result<(), LoopError> { + let destination = self.destination.as_str(); + let pool = self.native.pool.get()?; + let store = &self.native.store; + let reader = self.native.sync.reader(pool.clone()); + + let first_catch_up_check = queue.last_successful_stream_ordering().is_none(); + if first_catch_up_check { + let last = store + .get_destination_last_successful_stream_ordering(&pool, destination) + .await + .map_err(db_error)?; + queue.set_last_successful_stream_ordering(last); + } + let Some(mut last_successful) = queue.last_successful_stream_ordering() else { + // We have never sent this destination a PDU, so we do not know + // what it is missing. + queue.set_catching_up(false); + return Ok(()); + }; + + while queue.transmission_loop_enabled() { + let event_ids = store + .get_catch_up_room_event_ids(&pool, destination, last_successful) + .await + .map_err(db_error)?; + if event_ids.is_empty() { + // Check that no new event was skipped while we caught up. + let skipped = queue.catchup_last_skipped(); + if skipped != 0 && skipped > last_successful { + // Which is also how a room purged while we loop ends up, + // so reset the flag rather than loop hot. + queue.set_catchup_last_skipped(0); + continue; + } + queue.set_catching_up(false); + break; + } + + if first_catch_up_check { + // PDUs queued before we knew we had to catch up go. + queue.start_catching_up(); + } + + let ids: Vec> = event_ids.iter().map(|id| Arc::from(id.as_str())).collect(); + let pdus = reader + .events(&ids, RedactBehaviour::Redact, false, false) + .await + .map_err(db_error)?; + if pdus.is_empty() { + return Err(LoopError::Other(format!( + "No events retrieved when we asked for {event_ids:?}. This should not happen." + ))); + } + log::info!( + target: QUEUE_LOG, + "Catching up destination {destination} with {} PDUs", + pdus.len() + ); + + // One room per transaction, so the destination gets small + // amounts of work. There is one PDU per room. + for pdu in pdus { + let stream_ordering = stream_ordering_of(&pdu)?; + let room_id = pdu.room_id().to_owned(); + let room_pdus = self + .room_catch_up_pdus(&reader, &pool, pdu, last_successful) + .await?; + if !queue.transmission_loop_enabled() { + return Ok(()); + } + + log::info!(target: QUEUE_LOG, "Catching up rooms to {destination}: '{room_id}'"); + let sent_stream_orderings = room_pdus + .iter() + .map(stream_ordering_of) + .collect::, _>>()?; + let room_pdus = room_pdus + .into_iter() + .map(queued_pdu) + .collect::, _>>()?; + let sent = queue.send_pdus(observer, room_pdus).await?; + self.complete( + observer, + Completion { + to_device_stream_id: None, + device_list_stream_id: None, + last_stream_ordering: None, + sticky: None, + sent: Some(sent), + }, + ) + .await?; + + // Sticky events in the gap we skip over become a backlog. + store + .mark_backlogged_sticky_events_after_catchup_transaction( + &pool, + destination, + last_successful, + stream_ordering, + &sent_stream_orderings, + ) + .await + .map_err(db_error)?; + queue.set_sticky_backlogged(true); + + // Our place in the queue of missed PDUs is the original PDU, + // not those we sent. + last_successful = stream_ordering; + queue.set_last_successful_stream_ordering(Some(stream_ordering)); + store + .set_destination_last_successful_stream_ordering( + &pool, + destination, + stream_ordering, + ) + .await + .map_err(db_error)?; + } + } + Ok(()) } /// `get_new_device_msgs_for_remote` and `get_device_updates_by_remote`, diff --git a/rust/src/federation/queue/python.rs b/rust/src/federation/queue/python.rs index 43961b9535..a759e493c6 100644 --- a/rust/src/federation/queue/python.rs +++ b/rust/src/federation/queue/python.rs @@ -56,7 +56,11 @@ use crate::{ }, }, runtime::RustRuntime, - storage::federation_store::{FederationStore, FederationStoreConfig}, + homeserver::HomeServer, + storage::{ + federation_store::{FederationStore, FederationStoreConfig}, + sync_store::{SyncStore, SyncStoreConfig}, + }, }; /// Makes the queues, and tells tests when their loops are idle. @@ -105,22 +109,35 @@ impl RustDestinationQueues { /// `to_device_changed` and `device_list_changed`. /// /// Only the queues made after this use the native driver. + #[allow(clippy::too_many_arguments)] fn use_native_store( &self, + py: Python<'_>, + homeserver: HomeServer, db_pool: Py, stream_tokens: Py, allow_device_name_lookup_over_federation: bool, to_device_since: i64, device_lists_since: i64, ) -> PyResult<()> { + let config = homeserver.config(py)?; + let server_name = homeserver.hostname(py)?; + let store = homeserver.get_rust_store(py)?; + let sync = SyncStore::new( + Arc::clone(&store.caches), + SyncStoreConfig::from_config(&config), + ); let native = Arc::new(Native { store: FederationStore::new(FederationStoreConfig { allow_device_name_lookup_over_federation, + server_name: server_name.clone(), }), pool: NativePool::new(db_pool), stream_tokens, to_device: StreamChanges::new(to_device_since), device_lists: StreamChanges::new(device_lists_since), + sync: Arc::new(sync), + server_name, }); self.native.set(native).map_err(|_| { pyo3::exceptions::PyRuntimeError::new_err("the native store is already in use") @@ -469,7 +486,11 @@ impl PyHooks { #[async_trait] impl QueueHooks for PyHooks { - async fn catch_up(&self, observer: &Observer) -> Result<(), LoopError> { + async fn catch_up( + &self, + _queue: &DestinationQueue, + observer: &Observer, + ) -> Result<(), LoopError> { self.call( observer, "queue_catch_up", @@ -601,6 +622,7 @@ fn completion_args<'py>(py: Python<'py>, completion: &Completion) -> PyResult pdus.append(event.clone_ref(py))?, + PduBody::Native(event) => pdus.append(Py::new(py, event.shallow_clone())?)?, #[cfg(test)] PduBody::Json(_) => {} } diff --git a/rust/src/federation/queue/tests.rs b/rust/src/federation/queue/tests.rs index 56f0e3510d..3452a6f6b6 100644 --- a/rust/src/federation/queue/tests.rs +++ b/rust/src/federation/queue/tests.rs @@ -133,7 +133,11 @@ impl FakeHooks { #[async_trait] impl QueueHooks for FakeHooks { - async fn catch_up(&self, _observer: &Observer) -> Result<(), LoopError> { + async fn catch_up( + &self, + _queue: &DestinationQueue, + _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(); diff --git a/rust/src/storage/federation_store.rs b/rust/src/storage/federation_store.rs index be1d50c882..ebb07bc0ea 100644 --- a/rust/src/storage/federation_store.rs +++ b/rust/src/storage/federation_store.rs @@ -44,12 +44,25 @@ const SIGNING_KEY_UPDATE: &str = "m.signing_key_update"; const UNSTABLE_SIGNING_KEY_UPDATE: &str = "org.matrix.signing_key_update"; /// The parts of the homeserver config the queries apply. -#[derive(Debug, Clone, Copy, Default)] +#[derive(Debug, Clone, Default)] pub struct FederationStoreConfig { /// `allow_device_name_lookup_over_federation`. pub allow_device_name_lookup_over_federation: bool, + /// This server's name, for the sticky event queries, which only send + /// local users' events. + pub server_name: String, } +impl FederationStoreConfig { + /// `user_is_local_like_pattern`. + fn local_users_pattern(&self) -> String { + format!("@%:{}", self.server_name) + } +} + +/// How many catch-up events to read at a time. +const CATCH_UP_BATCH_SIZE: i64 = 50; + /// The transmission loop's queries. pub struct FederationStore { config: FederationStoreConfig, @@ -419,6 +432,122 @@ impl FederationStore { .await?; Ok(()) } + + /// `get_destination_last_successful_stream_ordering`: the stream + /// ordering of the last PDU sent to `destination`, if we know it. + pub async fn get_destination_last_successful_stream_ordering( + &self, + pool: &ConnectionPool, + destination: &str, + ) -> anyhow::Result> { + let rows = Conn::borrow(pool) + .await? + .query( + "SELECT last_successful_stream_ordering FROM destinations + WHERE destination = $1", + &[&destination], + ) + .await?; + match rows.first() { + Some(row) => Ok(row.try_get(0)?), + None => Ok(None), + } + } + + /// `get_catch_up_room_event_ids`: the newest event of each room we + /// share with `destination` that it has not been sent, oldest first, up + /// to 50. + pub async fn get_catch_up_room_event_ids( + &self, + pool: &ConnectionPool, + destination: &str, + last_successful_stream_ordering: i64, + ) -> anyhow::Result> { + let rows = Conn::borrow(pool) + .await? + .query( + "SELECT event_id FROM destination_rooms + JOIN events USING (stream_ordering) + WHERE destination = $1 AND stream_ordering > $2 + ORDER BY stream_ordering + LIMIT $3", + &[ + &destination, + &last_successful_stream_ordering, + &CATCH_UP_BATCH_SIZE, + ], + ) + .await?; + rows.iter() + .map(|row| Ok(row.try_get(0)?)) + .collect() + } + + /// `get_prev_events_for_room`: the room's 10 deepest forward + /// extremities. + pub async fn get_prev_events_for_room( + &self, + pool: &ConnectionPool, + room_id: &str, + ) -> anyhow::Result> { + let rows = Conn::borrow(pool) + .await? + .query( + "SELECT e.event_id FROM event_forward_extremities AS f + INNER JOIN events AS e USING (event_id) + WHERE f.room_id = $1 + ORDER BY e.depth DESC + LIMIT 10", + &[&room_id], + ) + .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 + /// events in that gap as backlogged, from its first such event, except + /// those just sent. + pub async fn mark_backlogged_sticky_events_after_catchup_transaction( + &self, + pool: &ConnectionPool, + destination: &str, + old_position: i64, + new_position: i64, + sent_stream_orderings: &[i64], + ) -> anyhow::Result<()> { + Conn::borrow(pool) + .await? + .query( + "INSERT INTO destination_rooms_sticky_events_backlog AS backlog + (destination, room_id, sticky_events_stream_position) + SELECT $1, dr.room_id, MIN(se.stream_id) + FROM destination_rooms AS dr + INNER JOIN sticky_events se USING (room_id) + WHERE dr.destination = $1 + AND $2::bigint < dr.stream_ordering + AND $2::bigint < se.event_stream_ordering + AND se.event_stream_ordering < $3::bigint + AND se.sender LIKE $4 + AND NOT (se.event_stream_ordering = ANY($5::bigint[])) + GROUP BY dr.room_id + ON CONFLICT (destination, room_id) + DO UPDATE SET sticky_events_stream_position = EXCLUDED.sticky_events_stream_position + WHERE EXCLUDED.sticky_events_stream_position < backlog.sticky_events_stream_position", + &[ + &destination, + &old_position, + &new_position, + &self.config.local_users_pattern(), + &sent_stream_orderings, + ], + ) + .await?; + Ok(()) + } } /// A row of `device_lists_outbound_pokes`. @@ -792,11 +921,13 @@ impl PyFederationStoreForTest { fn new( pool: &Bound<'_, PyConnectionPool>, allow_device_name_lookup_over_federation: bool, + server_name: String, ) -> Self { let pool = pool.get(); PyFederationStoreForTest { store: FederationStore::new(FederationStoreConfig { allow_device_name_lookup_over_federation, + server_name, }), pool: pool.pool.clone(), handle: pool.handle.clone(), @@ -891,6 +1022,59 @@ impl PyFederationStoreForTest { ), ) } + + fn get_destination_last_successful_stream_ordering( + &self, + py: Python<'_>, + destination: &str, + ) -> PyResult> { + self.run( + py, + self.store + .get_destination_last_successful_stream_ordering(&self.pool, destination), + ) + } + + fn get_catch_up_room_event_ids( + &self, + py: Python<'_>, + destination: &str, + last_successful_stream_ordering: i64, + ) -> PyResult> { + self.run( + py, + self.store.get_catch_up_room_event_ids( + &self.pool, + destination, + last_successful_stream_ordering, + ), + ) + } + + fn get_prev_events_for_room(&self, py: Python<'_>, room_id: &str) -> PyResult> { + self.run(py, self.store.get_prev_events_for_room(&self.pool, room_id)) + } + + fn mark_backlogged_sticky_events_after_catchup_transaction( + &self, + py: Python<'_>, + destination: &str, + old_position: i64, + new_position: i64, + sent_stream_orderings: Vec, + ) -> PyResult<()> { + self.run( + py, + self.store + .mark_backlogged_sticky_events_after_catchup_transaction( + &self.pool, + destination, + old_position, + new_position, + &sent_stream_orderings, + ), + ) + } } #[cfg(test)] diff --git a/synapse/federation/sender/rust_queue.py b/synapse/federation/sender/rust_queue.py index cd1d3773fc..92b153b8e5 100644 --- a/synapse/federation/sender/rust_queue.py +++ b/synapse/federation/sender/rust_queue.py @@ -83,6 +83,7 @@ def use_native_store( rust_queues.device_list_changed ) rust_queues.use_native_store( + hs, db_pool, lambda: ( store.get_to_device_stream_token(), diff --git a/synapse/synapse_rust/federation.pyi b/synapse/synapse_rust/federation.pyi index 2136b0da7d..d6e4f0e4aa 100644 --- a/synapse/synapse_rust/federation.pyi +++ b/synapse/synapse_rust/federation.pyi @@ -216,14 +216,15 @@ class RustDestinationQueues: def use_native_store( self, + homeserver: HomeServer, db_pool: Any, stream_tokens: Callable[[], tuple[int, int]], allow_device_name_lookup_over_federation: bool, to_device_since: int, device_lists_since: int, ) -> None: - """Do the loops' database work through the native driver. `db_pool` - is the `RustConnectionPool`, and `stream_tokens()` returns the current + """Do the loops' database work, and catch-up, through the native + driver. `db_pool` is the `RustConnectionPool`, and `stream_tokens()` returns the current to-device and device list stream positions. The change trackers know about the changes after the two positions given, and must be told of every later one.""" @@ -295,7 +296,10 @@ class FederationStoreForTest: Each method blocks until its query is done.""" def __init__( - self, pool: Any, allow_device_name_lookup_over_federation: bool + self, + pool: Any, + allow_device_name_lookup_over_federation: bool, + server_name: str, ) -> None: ... def get_new_device_msgs_for_remote( self, destination: str, last_stream_id: int, current_stream_id: int, limit: int @@ -312,3 +316,17 @@ class FederationStoreForTest: def set_destination_last_successful_stream_ordering( self, destination: str, stream_ordering: int ) -> None: ... + def get_destination_last_successful_stream_ordering( + self, destination: str + ) -> int | None: ... + def get_catch_up_room_event_ids( + self, destination: str, last_successful_stream_ordering: int + ) -> list[str]: ... + def get_prev_events_for_room(self, room_id: str) -> list[str]: ... + def mark_backlogged_sticky_events_after_catchup_transaction( + self, + destination: str, + old_position: int, + new_position: int, + sent_stream_orderings: list[int], + ) -> None: ... diff --git a/tests/federation/test_rust_queues.py b/tests/federation/test_rust_queues.py index ad9add63bc..d6156a8c39 100644 --- a/tests/federation/test_rust_queues.py +++ b/tests/federation/test_rust_queues.py @@ -321,6 +321,79 @@ class RustQueuesNativeTestCase(RustQueuesTestCase): self.assertEqual(len(self.transport.requests), 1) self.assertEqual(self._queue_calls(before), {}) + def _catch_up_destination(self, destination: str, missed: EventBase) -> None: + """Make `destination` one we sent nothing to since before `missed`.""" + stream_ordering = missed.internal_metadata.stream_ordering + assert stream_ordering is not None + self.get_success( + self.store.set_destination_last_successful_stream_ordering( + destination, stream_ordering - 1 + ) + ) + self.get_success( + self.store.store_destination_rooms_entries( + [destination], missed.room_id, stream_ordering + ) + ) + + def test_catch_up(self) -> None: + """Catch-up runs in Rust: it sends the room's latest event in place of + the one the destination missed, and records how far it got, without + taking the GIL.""" + self.hs.get_module_api_callbacks().federation._on_event_delivered_over_federation_callbacks.clear() + destination = "catchup.example.com" + self._catch_up_destination(destination, self.event) + latest = self._send_event("newer") + queue = self.sender._get_per_destination_queue(destination) + assert isinstance(queue, RustPerDestinationQueue) + self.transport.responses.append(json_response(200, {})) + + before = federation_gil_stats() + queue.attempt_new_transaction() + self._settle() + + [request] = self.transport.requests + body = json.loads(request.body) + self.assertEqual( + [pdu["content"].get("body") for pdu in body["pdus"]], + [latest.content["body"]], + ) + self.assertEqual( + self.get_success( + self.store.get_destination_last_successful_stream_ordering(destination) + ), + self.event.internal_metadata.stream_ordering, + ) + self.assertFalse(queue._catching_up) + self.assertNotIn("queue_catch_up", self._queue_calls(before)) + + def test_catch_up_falls_back(self) -> None: + """With history visible to members only, the destination, which has + no members, may not see the room's latest events, so catch-up sends + the event it missed.""" + destination = "catchup.example.com" + self._catch_up_destination(destination, self.event) + self.helper.send_state( + self.room_id, + "m.room.history_visibility", + {"history_visibility": "joined"}, + tok=self.tok, + ) + self._send_event("newer") + queue = self.sender._get_per_destination_queue(destination) + assert isinstance(queue, RustPerDestinationQueue) + self.transport.responses.append(json_response(200, {})) + + queue.attempt_new_transaction() + self._settle() + + [request] = self.transport.requests + body = json.loads(request.body) + self.assertEqual( + [pdu["content"].get("body") for pdu in body["pdus"]], + [self.event.content["body"]], + ) + 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 ecae32322a..002316958f 100644 --- a/tests/storage/test_federation_store.py +++ b/tests/storage/test_federation_store.py @@ -31,11 +31,14 @@ from signedjson.key import ( from twisted.internet.testing import MemoryReactor from synapse.handlers.e2e_keys import SignatureListItem +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.util.clock import Clock +from synapse.util.duration import Duration from tests import unittest @@ -76,8 +79,8 @@ class FederationStoreTestCase(unittest.HomeserverTestCase): db_pool = self.store.db_pool._db_pool assert isinstance(db_pool, RustConnectionPool) assert db_pool._pool is not None - self.rust = FederationStoreForTest(db_pool._pool, False) - self.rust_with_names = FederationStoreForTest(db_pool._pool, True) + self.rust = FederationStoreForTest(db_pool._pool, False, hs.hostname) + self.rust_with_names = FederationStoreForTest(db_pool._pool, True, hs.hostname) # To-device messages. @@ -365,3 +368,175 @@ class FederationStoreTestCase(unittest.HomeserverTestCase): ), 7, ) + + +class CatchUpStoreTestCase(unittest.HomeserverTestCase): + """The catch-up queries against Python's, on the same database.""" + + servlets = [ + admin.register_servlets, + login.register_servlets, + room.register_servlets, + ] + + 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() + if not isinstance( + hs.get_datastores().main.db_pool._db_pool, RustConnectionPool + ): + raise SkipTest("needs the native Rust Postgres driver") + return hs + + def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None: + self.store = hs.get_datastores().main + db_pool = self.store.db_pool._db_pool + assert isinstance(db_pool, RustConnectionPool) + assert db_pool._pool is not None + self.rust = FederationStoreForTest(db_pool._pool, False, hs.hostname) + + self.user = self.register_user("alice", "pass") + self.tok = self.login("alice", "pass") + self.rooms = [ + self.helper.create_room_as(self.user, tok=self.tok) for _ in range(3) + ] + + def _send(self, room_id: str, sticky: bool) -> int: + """Send a message, record it as due to both destinations, and return + its stream ordering.""" + content = {"body": "hi", "msgtype": "m.text"} + if sticky: + event_id = self.helper.send_sticky_event( + room_id, + "m.room.message", + duration=Duration(minutes=10), + content=content, + tok=self.tok, + )["event_id"] + else: + event_id = self.helper.send_event( + room_id, "m.room.message", content, tok=self.tok + )["event_id"] + event = self.get_success(self.store.get_event(event_id)) + stream_ordering = event.internal_metadata.stream_ordering + assert stream_ordering is not None + self.get_success( + self.store.store_destination_rooms_entries( + [DEST, OTHER], room_id, stream_ordering + ) + ) + return stream_ordering + + def test_last_successful_stream_ordering(self) -> None: + self.assertIsNone( + self.rust.get_destination_last_successful_stream_ordering(DEST) + ) + self.get_success( + self.store.set_destination_last_successful_stream_ordering(DEST, 9) + ) + self.assertEqual( + self.rust.get_destination_last_successful_stream_ordering(DEST), 9 + ) + + def test_catch_up_room_event_ids(self) -> None: + orderings = [ + self._send(room_id, sticky=False) + for room_id in self.rooms + for _ in range(2) + ] + for last in [0, *orderings]: + for destination in (DEST, OTHER, "unknown"): + self.assertEqual( + self.rust.get_catch_up_room_event_ids(destination, last), + self.get_success( + self.store.get_catch_up_room_event_ids(destination, last) + ), + ) + + def test_prev_events_for_room(self) -> None: + for room_id in self.rooms: + self._send(room_id, sticky=False) + for room_id in [*self.rooms, "!unknown:test"]: + self.assertEqual( + self.rust.get_prev_events_for_room(room_id), + self.get_success(self.store.get_prev_events_for_room(room_id)), + ) + + def _backlog(self) -> list[tuple[str, str, int]]: + rows = self.get_success( + self.store.db_pool.simple_select_list( + "destination_rooms_sticky_events_backlog", + None, + ["destination", "room_id", "sticky_events_stream_position"], + ) + ) + return sorted(rows) + + def _clear_backlog(self) -> None: + self.get_success( + self.store.db_pool.runInteraction( + "clear_backlog", + lambda txn: txn.execute( + "DELETE FROM destination_rooms_sticky_events_backlog" + ), + ) + ) + + def test_mark_backlogged_sticky_events_after_catchup_transaction(self) -> None: + orderings = [] + for i in range(9): + orderings.append(self._send(self.rooms[i % 3], sticky=i % 2 == 0)) + + cases = [ + (0, orderings[-1] + 1, []), + (0, orderings[-1] + 1, orderings[::3]), + (orderings[2], orderings[7], [orderings[4]]), + (orderings[5], orderings[6], []), + ] + for old, new, sent in cases: + # With and without a backlog already further on. + for existing in (None, 1_000_000): + self._clear_backlog() + if existing is not None: + self.get_success( + self.store.db_pool.simple_insert( + "destination_rooms_sticky_events_backlog", + { + "destination": DEST, + "room_id": self.rooms[0], + "sticky_events_stream_position": existing, + }, + ) + ) + self.get_success( + self.store.mark_backlogged_sticky_events_after_catchup_transaction( + DEST, + old_last_successfully_sent_stream_ordering=old, + new_last_successfully_sent_stream_ordering=new, + event_stream_orderings_sent_in_transaction=sent, + ) + ) + python = self._backlog() + + self._clear_backlog() + if existing is not None: + self.get_success( + self.store.db_pool.simple_insert( + "destination_rooms_sticky_events_backlog", + { + "destination": DEST, + "room_id": self.rooms[0], + "sticky_events_stream_position": existing, + }, + ) + ) + self.rust.mark_backlogged_sticky_events_after_catchup_transaction( + DEST, old, new, sent + ) + self.assertEqual(self._backlog(), python, (old, new, sent, existing)) + # The cases did backlog something. + self.assertNotEqual(python, [])