diff --git a/bindings/matrix-sdk-ffi/src/client.rs b/bindings/matrix-sdk-ffi/src/client.rs index b4c52ab8c..c122cf5d3 100644 --- a/bindings/matrix-sdk-ffi/src/client.rs +++ b/bindings/matrix-sdk-ffi/src/client.rs @@ -2155,6 +2155,15 @@ impl Client { })))) } + /// Whether to enable automatic backpagination under certain conditions + /// (e.g. when processing read receipts). + /// + /// This is an experimental feature, and might cause performance issues on + /// large accounts. Use with caution. + pub fn enable_automatic_backpagination(&self) { + self.inner.event_cache().config_mut().experimental_auto_backpagination = true; + } + pub fn homeserver_capabilities(&self) -> HomeserverCapabilities { HomeserverCapabilities::new(self.inner.homeserver_capabilities()) } diff --git a/crates/matrix-sdk-ui/tests/integration/timeline/pinned_event.rs b/crates/matrix-sdk-ui/tests/integration/timeline/pinned_event.rs index b93eee4ed..5512641f9 100644 --- a/crates/matrix-sdk-ui/tests/integration/timeline/pinned_event.rs +++ b/crates/matrix-sdk-ui/tests/integration/timeline/pinned_event.rs @@ -389,7 +389,7 @@ async fn test_max_events_to_load_is_honored() { let client = server.client_builder().build().await; let room_id = room_id!("!test:localhost"); - client.event_cache().config_mut().await.max_pinned_events_to_load = 1; + client.event_cache().config_mut().max_pinned_events_to_load = 1; let f = EventFactory::new().room(room_id).sender(*BOB); let pinned_event = f @@ -869,7 +869,7 @@ async fn test_ensure_max_concurrency_is_observed() { let max_concurrent_requests = 10; // Define the max concurrent requests allowed for the event cache. - client.event_cache().config_mut().await.max_pinned_events_concurrent_requests = + client.event_cache().config_mut().max_pinned_events_concurrent_requests = max_concurrent_requests; let f = EventFactory::new().room(&room_id).sender(user_id!("@example:localhost")); diff --git a/crates/matrix-sdk/src/event_cache/automatic_pagination.rs b/crates/matrix-sdk/src/event_cache/automatic_pagination.rs new file mode 100644 index 000000000..51c62cfa6 --- /dev/null +++ b/crates/matrix-sdk/src/event_cache/automatic_pagination.rs @@ -0,0 +1,347 @@ +// Copyright 2026 The Matrix.org Foundation C.I.C. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::{ + collections::HashMap, + sync::{Arc, Weak}, +}; + +use matrix_sdk_base::task_monitor::{BackgroundTaskHandle, TaskMonitor}; +use ruma::{OwnedRoomId, RoomId}; +use tokio::sync::mpsc; +use tracing::{info, instrument, trace, warn}; + +use crate::event_cache::EventCacheInner; + +/// State for running paginations in background tasks. +/// +/// Shallow type, can be cloned cheaply. +#[derive(Clone)] +pub struct AutomaticPagination { + inner: Arc, +} + +#[cfg(not(tarpaulin_include))] +impl std::fmt::Debug for AutomaticPagination { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("AutomaticPagination").finish_non_exhaustive() + } +} + +impl AutomaticPagination { + /// Create a new [`AutomaticPagination`], spawning the background task to + /// handle incoming requests to run background paginations. + pub(super) fn new(event_cache: Weak, task_monitor: &TaskMonitor) -> Self { + let (sender, receiver) = mpsc::unbounded_channel(); + + let task = task_monitor.spawn_background_task( + "event_cache::automatic_paginations_task", + automatic_paginations_task(event_cache, receiver), + ); + + Self { inner: Arc::new(AutomaticPaginationInner { _task: task, sender }) } + } + + /// Request a single back-pagination to happen in the background for the + /// given room. + /// + /// Returns false, if the request couldn't be sent. + pub fn run_once(&self, room_id: &RoomId) -> bool { + // We don't want to do anything with the error type, as it only includes the + // request we just created, and not much more; there's no guarantee that + // retrying sending it would succeed, so let it drop, and report the + // result as a boolean, for informative purposes. + self.inner + .sender + .send(AutomaticPaginationRequest::PaginateRoomBackwards { room_id: room_id.to_owned() }) + .is_ok() + } +} + +struct AutomaticPaginationInner { + /// The task used to handle automatic pagination requests. + _task: BackgroundTaskHandle, + + /// A sender for automatic pagination requests, that is shared with every + /// room. + /// + /// It's a `OnceLock` because its initialization is deferred to + /// [`EventCache::subscribe`]. + sender: mpsc::UnboundedSender, +} + +#[derive(Clone, Debug)] +enum AutomaticPaginationRequest { + PaginateRoomBackwards { room_id: OwnedRoomId }, +} + +/// Listen to background automatic pagination requests, and execute them in +/// real-time. +#[instrument(skip_all)] +async fn automatic_paginations_task( + inner: Weak, + mut receiver: mpsc::UnboundedReceiver, +) { + trace!("Spawning the automatic pagination task"); + + let mut room_pagination_credits = HashMap::new(); + + while let Some(request) = receiver.recv().await { + match request { + AutomaticPaginationRequest::PaginateRoomBackwards { room_id } => { + let Some(inner) = inner.upgrade() else { + // The event cache has been dropped, exit the task. + break; + }; + + let config = *inner.config.read().unwrap(); + + let credits = room_pagination_credits + .entry(room_id.clone()) + .or_insert(config.room_pagination_per_room_credit); + + if *credits == 0 { + trace!(for_room = %room_id, "No more credits to paginate this room in the background, skipping"); + continue; + } + + let pagination = match inner.all_caches_for_room(&room_id).await { + Ok(caches) => caches.room.pagination(), + Err(err) => { + warn!(for_room = %room_id, "Failed to get the `Caches`: {err}"); + continue; + } + }; + + trace!(for_room = %room_id, "automatic backpagination triggered"); + + match pagination.run_backwards_once(config.room_pagination_batch_size).await { + Ok(outcome) => { + // Pagination requests must be idempotent, so we only decrement credits if + // we actually paginated something new. + if !outcome.reached_start || !outcome.events.is_empty() { + *credits -= 1; + } + } + + Err(err) => { + warn!(for_room = %room_id, "Failed to run background pagination: {err}"); + // Don't decrement credits in this case, to allow a + // retry later. + } + } + } + } + } + + // The sender has shut down, exit. + info!("Closing the automatic pagination task because receiver closed"); +} + +// MatrixMockServer et al. aren't available on wasm. +#[cfg(all(test, not(target_arch = "wasm32")))] +mod tests { + use std::time::Duration; + + use assert_matches::assert_matches; + use eyeball_im::VectorDiff; + use matrix_sdk_base::sleep::sleep; + use matrix_sdk_test::{BOB, JoinedRoomBuilder, async_test, event_factory::EventFactory}; + use ruma::{event_id, room_id}; + + use crate::{ + assert_let_timeout, + event_cache::{EventsOrigin, RoomEventCacheUpdate}, + test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate}, + }; + + /// Test that we can send automatic pagination requests. + #[async_test] + async fn test_background_room_paginations() { + let server = MatrixMockServer::new().await; + let client = server.client_builder().build().await; + + let event_cache = client.event_cache(); + event_cache.config_mut().experimental_auto_backpagination = true; + event_cache.subscribe().unwrap(); + + let room_id = room_id!("!omelette:fromage.fr"); + let f = EventFactory::new().room(room_id).sender(*BOB); + + let room = server.sync_joined_room(&client, room_id).await; + + let (room_event_cache, _drop_handles) = room.event_cache().await.unwrap(); + + // Starting with an empty, inactive room, + let (room_events, mut room_cache_updates) = room_event_cache.subscribe().await.unwrap(); + assert!(room_events.is_empty()); + assert!(room_cache_updates.is_empty()); + + // We get a gappy sync (so as to have a previous-batch token), + server + .sync_room( + &client, + JoinedRoomBuilder::new(room_id) + .set_timeline_limited() + .set_timeline_prev_batch("prev_batch"), + ) + .await; + + { + assert_let_timeout!( + Ok(RoomEventCacheUpdate::UpdateTimelineEvents(update)) = room_cache_updates.recv() + ); + assert_eq!(update.diffs.len(), 1); + assert_matches!(update.diffs[0], VectorDiff::Clear); + assert_matches!(update.origin, EventsOrigin::Sync); + } + + // Set up the mock for /messages, + server + .mock_room_messages() + .ok(RoomMessagesResponseTemplate::default().events(vec![ + f.text_msg("comté").event_id(event_id!("$2")), + f.text_msg("beaufort").event_id(event_id!("$1")), + ])) + .mock_once() + .mount() + .await; + + // Send a request for a background pagination, + let automatic_pagination_api = event_cache.automatic_pagination().unwrap(); + assert!(automatic_pagination_api.run_once(room_id)); + + // The room pagination happens in the background. + assert_let_timeout!( + Ok(RoomEventCacheUpdate::UpdateTimelineEvents(update)) = room_cache_updates.recv() + ); + assert_eq!(update.diffs.len(), 1); + + assert_matches!(update.origin, EventsOrigin::Pagination); + + let mut room_events = room_events.into(); + for diff in update.diffs { + diff.apply(&mut room_events); + } + + assert_eq!(room_events.len(), 2); + assert_eq!(room_events[0].event_id().unwrap(), event_id!("$1")); + assert_eq!(room_events[1].event_id().unwrap(), event_id!("$2")); + + // And there's no more updates. + assert!(room_cache_updates.is_empty()); + } + + /// Test that the credit system works. + #[async_test] + async fn test_room_pagination_respects_credits_system() { + let server = MatrixMockServer::new().await; + let client = server.client_builder().build().await; + + let event_cache = client.event_cache(); + event_cache.config_mut().experimental_auto_backpagination = true; + + // Only allow 1 background pagination per room, to test that the credit system + // is properly taken into account. + event_cache.config_mut().room_pagination_per_room_credit = 1; + event_cache.subscribe().unwrap(); + + let room_id = room_id!("!omelette:fromage.fr"); + let f = EventFactory::new().room(room_id).sender(*BOB); + + let room = server.sync_joined_room(&client, room_id).await; + let (room_event_cache, _drop_handles) = room.event_cache().await.unwrap(); + + // Starting with an empty, inactive room, + let (room_events, mut room_cache_updates) = room_event_cache.subscribe().await.unwrap(); + assert!(room_events.is_empty()); + assert!(room_cache_updates.is_empty()); + + server + .sync_room( + &client, + JoinedRoomBuilder::new(room_id) + .set_timeline_limited() + .set_timeline_prev_batch("prev_batch"), + ) + .await; + + { + assert_let_timeout!( + Ok(RoomEventCacheUpdate::UpdateTimelineEvents(update)) = room_cache_updates.recv() + ); + assert_eq!(update.diffs.len(), 1); + assert_matches!(update.diffs[0], VectorDiff::Clear); + assert_matches!(update.origin, EventsOrigin::Sync); + } + + // Set up the mock for /messages, so that it returns another prev-batch token, + server + .mock_room_messages() + .match_from("prev_batch") + .ok(RoomMessagesResponseTemplate::default() + .events(vec![ + f.text_msg("comté").event_id(event_id!("$2")), + f.text_msg("beaufort").event_id(event_id!("$1")), + ]) + .end_token("prev_batch_2")) + .mock_once() + .mount() + .await; + + // Send a request for a background pagination, + let automatic_pagination_api = event_cache.automatic_pagination().unwrap(); + assert!(automatic_pagination_api.run_once(room_id)); + + // The room pagination happens in the background. + assert_let_timeout!( + Ok(RoomEventCacheUpdate::UpdateTimelineEvents(update)) = room_cache_updates.recv() + ); + assert_eq!(update.diffs.len(), 1); + + assert_matches!(update.origin, EventsOrigin::Pagination); + + let mut room_events = room_events.into(); + for diff in update.diffs { + diff.apply(&mut room_events); + } + + assert_eq!(room_events.len(), 2); + assert_eq!(room_events[0].event_id().unwrap(), event_id!("$1")); + assert_eq!(room_events[1].event_id().unwrap(), event_id!("$2")); + + // And there's no more updates yet. + assert!(room_cache_updates.is_empty()); + + // One can send another request to back-paginate… + assert!(automatic_pagination_api.run_once(room_id)); + + sleep(Duration::from_millis(300)).await; + // But it doesn't happen, because we don't have enough credits for automatic + // backpagination. + assert!(room_cache_updates.is_empty()); + + // We can still manually backpaginate with success, though. + server + .mock_room_messages() + .match_from("prev_batch_2") + .ok(RoomMessagesResponseTemplate::default()) + .mock_once() + .mount() + .await; + + let outcome = room_event_cache.pagination().run_backwards_once(30).await.unwrap(); + assert!(outcome.reached_start); + } +} diff --git a/crates/matrix-sdk/src/event_cache/caches/mod.rs b/crates/matrix-sdk/src/event_cache/caches/mod.rs index 90d494590..3c15a7149 100644 --- a/crates/matrix-sdk/src/event_cache/caches/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/mod.rs @@ -24,7 +24,9 @@ use ruma::{OwnedRoomId, RoomId}; use tokio::sync::{broadcast::Sender, mpsc}; use super::{EventCacheError, EventsOrigin, Result}; -use crate::{client::WeakClient, room::WeakRoom}; +use crate::{ + client::WeakClient, event_cache::automatic_pagination::AutomaticPagination, room::WeakRoom, +}; pub mod event_focused; pub mod event_linked_chunk; @@ -50,6 +52,7 @@ impl Caches { linked_chunk_update_sender: Sender, auto_shrink_sender: mpsc::Sender, store: EventCacheStoreLock, + automatic_pagination: Option, ) -> Result { let Some(client) = weak_client.get() else { return Err(EventCacheError::ClientDropped); @@ -84,6 +87,7 @@ impl Caches { linked_chunk_update_sender, store, pagination_status.clone(), + automatic_pagination, ) .await?; diff --git a/crates/matrix-sdk/src/event_cache/caches/pinned_events/mod.rs b/crates/matrix-sdk/src/event_cache/caches/pinned_events/mod.rs index 0af9f9c35..819aba644 100644 --- a/crates/matrix-sdk/src/event_cache/caches/pinned_events/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/pinned_events/mod.rs @@ -463,7 +463,7 @@ impl PinnedEventCache { async fn reload_pinned_events(room: Room) -> Result>> { let (max_events_to_load, max_concurrent_requests) = { let client = room.client(); - let config = client.event_cache().config().await; + let config = client.event_cache().config(); (config.max_pinned_events_to_load, config.max_pinned_events_concurrent_requests) }; diff --git a/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs b/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs index 7da8c6c3e..9298070ef 100644 --- a/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs +++ b/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs @@ -110,7 +110,9 @@ use ruma::{ }; use tracing::{debug, instrument, trace, warn}; -use crate::event_cache::caches::event_linked_chunk::EventLinkedChunk; +use crate::event_cache::{ + automatic_pagination::AutomaticPagination, caches::event_linked_chunk::EventLinkedChunk, +}; trait RoomReadReceiptsExt { /// Update the [`RoomReadReceipts`] unread counts according to the new @@ -219,6 +221,20 @@ impl RoomReadReceiptsExt for RoomReadReceipts { } } +struct SelectBestReceiptResult { + /// Whether to request a back-pagination after trying to select the best + /// receipt. + /// + /// This is usually a signal that we didn't find any better receipt, or the + /// latest active receipt, in the current in-memory linked chunk, and + /// that we need to fetch more historical data (from disk or from the + /// network) to get a better picture of the unread counts. + request_pagination: bool, + + /// The event id of the best receipt we found, if any. + receipt: Option, +} + /// Return a new better (i.e. more recent) receipt based on a search of the /// linked chunk. /// @@ -235,7 +251,7 @@ fn select_best_receipt( new_receipt_event: Option<&ReceiptEventContent>, latest_active: Option<&EventId>, with_threading_support: bool, -) -> Option { +) -> SelectBestReceiptResult { // If we had a new receipt event, add the main/unthreaded receipts it contains // to the pending receipts list. We'll try to chase them later. if let Some(receipt_event) = new_receipt_event { @@ -263,17 +279,17 @@ fn select_best_receipt( // i.e., we've found a better receipt, *and* there's no more pending receipt // to try to match against events in the linked chunk. - let mut found = None; + let mut result = SelectBestReceiptResult { request_pagination: false, receipt: None }; for (event, event_id) in linked_chunk.revents().filter_map(|(_pos, ev)| Some((ev, ev.event_id()?))) { - if found.is_none() { + if result.receipt.is_none() { // Try to see if the latest active receipt is still the most recent receipt. if latest_active == Some(&event_id) { // The latest active receipt is still the most recent receipt, so keep it. trace!(active = %event_id, "the latest active receipt is still the most recent; stopping search"); - found = Some(event_id.clone()); + result.receipt = Some(event_id.clone()); } // Try to find an implicit read receipt (i.e. an event sent by the current // user). @@ -284,14 +300,14 @@ fn select_best_receipt( && (!with_threading_support || extract_thread_root(event.raw()).is_none()) { trace!(implicit = %event_id, "found an implicit receipt; stopping search"); - found = Some(event_id.clone()); + result.receipt = Some(event_id.clone()); } } // Early exit condition (see the comment above): we've already found a most // recent receipt, and there's no other pending receipts to match against known // events. - if found.is_some() && pending_receipts.is_empty() { + if result.receipt.is_some() && pending_receipts.is_empty() { trace!("exiting loop; found a better receipt, and no more pending receipt to match"); break; } @@ -301,9 +317,9 @@ fn select_best_receipt( // one! pending_receipts.retain(|pending| { if *pending == event_id { - if found.is_none() { + if result.receipt.is_none() { trace!(pending = %event_id, "found a pending receipt; stopping search"); - found = Some(event_id.clone()); + result.receipt = Some(event_id.clone()); } else { trace!(%event_id, "discarding a pending receipt that wasn't selected"); } @@ -318,13 +334,22 @@ fn select_best_receipt( }); } - found.or_else(|| latest_active.map(|event_id| { - // The only time when we can run into this is because we didn't find a more recent receipt, - // and the previously active one wasn't loaded in the linked chunk. In this case, every - // event currently loaded in the linked chunk will be processed in `process_event()`. - trace!(%event_id, "reusing previous active receipt (but we didn't find it in the linked chunk)"); - event_id.to_owned() - })) + if result.receipt.is_none() { + // Request a patination: we haven't found a better receipt, but we haven't even + // found the latest active receipt! + result.request_pagination = true; + + // In the meanwhile, reuse the latest active receipt, if any. + result.receipt = latest_active.map(|event_id| { + // The only time when we can run into this is because we didn't find a more recent receipt, + // and the previously active one wasn't loaded in the linked chunk. In this case, every + // event currently loaded in the linked chunk will be processed in `process_event()`. + trace!(%event_id, "reusing previous active receipt (but we didn't find it in the linked chunk)"); + event_id.to_owned() + }); + } + + result } /// Given a set of events coming from sync, for a room, update the @@ -340,30 +365,37 @@ pub(crate) fn compute_unread_counts( linked_chunk: &EventLinkedChunk, read_receipts: &mut RoomReadReceipts, with_threading_support: bool, + automatic_pagination: Option<&AutomaticPagination>, ) { debug!(?read_receipts, "Starting"); - let new_receipt = select_best_receipt( + let select_best_receipt_result = select_best_receipt( user_id, linked_chunk, &mut read_receipts.pending, receipt_event, read_receipts.latest_active.as_ref().map(|latest_active| latest_active.event_id.as_ref()), with_threading_support, - ) - .map(|event_id| LatestReadReceipt { event_id }); + ); - if let Some(new_receipt) = new_receipt { + if select_best_receipt_result.request_pagination { + trace!("Requesting pagination to find a better receipt"); + if let Some(automatic_pagination) = automatic_pagination + && !automatic_pagination.run_once(room_id) + { + warn!("Failed to request pagination to find a better receipt"); + } + } + + if let Some(event_id) = select_best_receipt_result.receipt { // We've found the id of an event to which the receipt attaches. The associated // event may either come from the new batch of events associated to // this sync, or it may live in the past timeline events we know // about. - let event_id = new_receipt.event_id.clone(); - // First, save the event id as the latest one that has a read receipt. trace!(%event_id, "Saving a new active read receipt"); - read_receipts.latest_active = Some(new_receipt); + read_receipts.latest_active = Some(LatestReadReceipt { event_id: event_id.clone() }); // The event for the receipt is in the linked chunk, so we'll find it and can // count safely from here. @@ -844,9 +876,11 @@ mod tests { active_receipt, with_threading_support, ); - assert!(result.is_none()); + assert!(result.receipt.is_none()); // And there are no pending receipts. assert!(pending_receipts.is_empty()); + // In this case, select_best_receipt will request a pagination. + assert!(result.request_pagination); } #[test] @@ -881,9 +915,11 @@ mod tests { active_receipt, with_threading_support, ); - assert_eq!(result.unwrap(), event_id!("$2")); + assert_eq!(result.receipt.unwrap(), event_id!("$2")); // And there are no pending receipts. assert!(pending_receipts.is_empty()); + // We don't need to paginate, as we found a receipt in the linked chunk. + assert!(result.request_pagination.not()); } #[test] @@ -918,9 +954,11 @@ mod tests { active_receipt, with_threading_support, ); - assert_eq!(result.unwrap(), event_id!("$2")); + assert_eq!(result.receipt.unwrap(), event_id!("$2")); // And there are no pending receipts. assert!(pending_receipts.is_empty()); + // We don't need to paginate, as we found a receipt in the linked chunk. + assert!(result.request_pagination.not()); } #[test] @@ -960,9 +998,11 @@ mod tests { active_receipt, with_threading_support, ); - assert_eq!(result.unwrap(), event_id!("$2")); + assert_eq!(result.receipt.unwrap(), event_id!("$2")); // And there are no pending receipts. assert!(pending_receipts.is_empty()); + // We don't need to paginate, as we found a receipt in the linked chunk. + assert!(result.request_pagination.not()); } #[test] @@ -1003,10 +1043,13 @@ mod tests { with_threading_support, ); - assert!(result.is_none()); + assert!(result.receipt.is_none()); // And there's a new pending receipt for $4. assert_eq!(pending_receipts.len(), 1); assert_eq!(pending_receipts.get(0).unwrap(), event_id!("$4")); + // We would need to paginate, as we haven't found a better receipt in the linked + // chunk. + assert!(result.request_pagination); } #[test] @@ -1043,9 +1086,11 @@ mod tests { active_receipt, with_threading_support, ); - assert_eq!(result.unwrap(), event_id!("$2")); + assert_eq!(result.receipt.unwrap(), event_id!("$2")); // And there are no more pending receipts. assert!(pending_receipts.is_empty()); + // We don't need to paginate, as we found a receipt in the linked chunk. + assert!(result.request_pagination.not()); } #[test] @@ -1093,12 +1138,15 @@ mod tests { active_receipt, with_threading_support, ); - assert_eq!(result.unwrap(), event_id!("$4")); + assert_eq!(result.receipt.unwrap(), event_id!("$4")); // Receipt 6 is still pending, and there's a new pending receipt for 7 too. ($2 // has been cleaned because it has been seen). assert_eq!(pending_receipts.len(), 2); assert!(pending_receipts.iter().any(|ev| ev == event_id!("$6"))); assert!(pending_receipts.iter().any(|ev| ev == event_id!("$7"))); + + // We don't need to paginate, as we found a receipt in the linked chunk. + assert!(result.request_pagination.not()); } } diff --git a/crates/matrix-sdk/src/event_cache/caches/room/state.rs b/crates/matrix-sdk/src/event_cache/caches/room/state.rs index 4a998c93a..ba2cb4c91 100644 --- a/crates/matrix-sdk/src/event_cache/caches/room/state.rs +++ b/crates/matrix-sdk/src/event_cache/caches/room/state.rs @@ -71,7 +71,13 @@ use super::{ RoomEventCacheLinkedChunkUpdate, RoomEventCacheUpdate, RoomEventCacheUpdateSender, sort_positions_descending, }; -use crate::{Room, event_cache::caches::pagination::SharedPaginationStatus, room::WeakRoom}; +use crate::{ + Room, + event_cache::{ + automatic_pagination::AutomaticPagination, caches::pagination::SharedPaginationStatus, + }, + room::WeakRoom, +}; /// Key for the event-focused caches. #[derive(Hash, PartialEq, Eq)] @@ -141,6 +147,9 @@ pub struct RoomEventCacheState { /// An atomic count of the current number of subscriber of the /// [`super::RoomEventCache`]. subscriber_count: Arc, + + /// A copy of the automatic pagination API object. + automatic_pagination: Option, } impl RoomEventCacheState { @@ -301,6 +310,7 @@ impl LockedRoomEventCacheState { linked_chunk_update_sender: Sender, store: EventCacheStoreLock, pagination_status: SharedObservable, + automatic_pagination: Option, ) -> Result { let store_guard = match store.lock().await? { // Lock is clean: all good! @@ -383,6 +393,7 @@ impl LockedRoomEventCacheState { waited_for_initial_prev_token: false, subscriber_count: Default::default(), pinned_event_cache: OnceLock::new(), + automatic_pagination, })) } } @@ -1025,9 +1036,8 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> { return Ok(()); }; - let state = &mut *self.state; - let user_id = &state.own_user_id; - let room_id = &state.room_id; + let user_id = &self.state.own_user_id; + let room_id = &self.state.room_id; let prev_read_receipts = room.read_receipts().clone(); let mut read_receipts = prev_read_receipts.clone(); @@ -1036,9 +1046,10 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> { user_id, room_id, receipt_event, - &state.room_linked_chunk, + &self.state.room_linked_chunk, &mut read_receipts, - state.enabled_thread_support, + self.state.enabled_thread_support, + self.state.automatic_pagination.as_ref(), ); if prev_read_receipts != read_receipts { diff --git a/crates/matrix-sdk/src/event_cache/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index a46d62c57..2dedeb5a0 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -30,8 +30,7 @@ use std::{ collections::HashMap, fmt, - ops::{Deref, DerefMut}, - sync::{Arc, OnceLock}, + sync::{Arc, OnceLock, RwLock as StdRwLock, RwLockReadGuard, RwLockWriteGuard}, }; use futures_util::future::try_join_all; @@ -57,6 +56,7 @@ use crate::{ paginators::PaginatorError, }; +mod automatic_pagination; mod caches; mod deduplicator; mod persistence; @@ -78,6 +78,8 @@ pub use caches::{ #[cfg(feature = "e2e-encryption")] pub use redecryptor::{DecryptionRetryRequest, RedecryptorReport}; +pub use crate::event_cache::automatic_pagination::AutomaticPagination; + /// An error observed in the [`EventCache`]. #[derive(thiserror::Error, Clone, Debug)] pub enum EventCacheError { @@ -214,7 +216,7 @@ impl EventCache { Self { inner: Arc::new(EventCacheInner { client: weak_client, - config: RwLock::new(EventCacheConfig::default()), + config: StdRwLock::new(EventCacheConfig::default()), store: event_cache_store, multiple_room_updates_lock: Default::default(), by_room: Default::default(), @@ -224,6 +226,7 @@ impl EventCache { linked_chunk_update_sender, #[cfg(feature = "e2e-encryption")] redecryption_channels, + automatic_pagination: OnceLock::new(), thread_subscriber_sender, }), } @@ -231,13 +234,13 @@ impl EventCache { /// Get a read-only handle to the global configuration of the /// [`EventCache`]. - pub async fn config(&self) -> impl Deref + '_ { - self.inner.config.read().await + pub fn config(&self) -> RwLockReadGuard<'_, EventCacheConfig> { + self.inner.config.read().unwrap() } /// Get a writable handle to the global configuration of the [`EventCache`]. - pub async fn config_mut(&self) -> impl DerefMut + '_ { - self.inner.config.write().await + pub fn config_mut(&self) -> RwLockWriteGuard<'_, EventCacheConfig> { + self.inner.config.write().unwrap() } /// Subscribes to updates that a thread subscription has been sent. @@ -318,6 +321,15 @@ impl EventCache { ) .abort_on_drop(); + if self.config().experimental_auto_backpagination { + // Run the deferred initialization of the automatic pagination request sender, that + // is shared with every room. + trace!("spawning the automatic paginations API"); + self.inner.automatic_pagination.get_or_init(|| AutomaticPagination::new(Arc::downgrade(&self.inner), task_monitor)); + } else { + trace!("automatic paginations API is disabled"); + } + Arc::new(EventCacheDropHandles { _listen_updates_task: listen_updates_task, _ignore_user_list_update_task: ignore_user_list_update_task, @@ -377,6 +389,13 @@ impl EventCache { pub fn subscribe_to_room_generic_updates(&self) -> Receiver { self.inner.generic_update_sender.subscribe() } + + /// Returns a reference to the [`AutomaticPagination`] API, if enabled at + /// construction with the + /// [`EventCacheConfig::experimental_auto_backpagination`] flag. + pub fn automatic_pagination(&self) -> Option { + self.inner.automatic_pagination.get().cloned() + } } /// Global configuration for the [`EventCache`], applied to every single room. @@ -387,15 +406,46 @@ pub struct EventCacheConfig { /// Maximum number of pinned events to load, for any room. pub max_pinned_events_to_load: usize, + + /// Whether to automatically backpaginate a room under certain conditions. + /// + /// Off by default. + pub experimental_auto_backpagination: bool, + + /// The maximum number of allowed room paginations, for a given room, that + /// can be executed in the automatic paginations task. + /// + /// After that number of paginations, the task will stop executing + /// paginations for that room *in the background* (user-requested + /// paginations will still be executed, of course). + /// + /// Defaults to [`EventCacheConfig::DEFAULT_ROOM_PAGINATION_CREDITS`]. + pub room_pagination_per_room_credit: usize, + + /// The number of messages to paginate in a single batch, when executing an + /// automatic pagination request. + /// + /// Defaults to [`EventCacheConfig::DEFAULT_ROOM_PAGINATION_BATCH_SIZE`]. + pub room_pagination_batch_size: u16, } impl EventCacheConfig { /// The default maximum number of pinned events to load. - const DEFAULT_MAX_EVENTS_TO_LOAD: usize = 128; + pub const DEFAULT_MAX_EVENTS_TO_LOAD: usize = 128; /// The default maximum number of concurrent requests to perform when /// loading the pinned events. - const DEFAULT_MAX_CONCURRENT_REQUESTS: usize = 8; + pub const DEFAULT_MAX_CONCURRENT_REQUESTS: usize = 8; + + /// The default number of credits to give to a room for automatic + /// paginations (see also + /// [`EventCacheConfig::room_pagination_per_room_credit`]). + pub const DEFAULT_ROOM_PAGINATION_CREDITS: usize = 20; + + /// The default number of messages to paginate in a single batch, when + /// executing an automatic pagination request (see also + /// [`EventCacheConfig::room_pagination_batch_size`]). + pub const DEFAULT_ROOM_PAGINATION_BATCH_SIZE: u16 = 30; } impl Default for EventCacheConfig { @@ -403,6 +453,9 @@ impl Default for EventCacheConfig { Self { max_pinned_events_concurrent_requests: Self::DEFAULT_MAX_CONCURRENT_REQUESTS, max_pinned_events_to_load: Self::DEFAULT_MAX_EVENTS_TO_LOAD, + room_pagination_per_room_credit: Self::DEFAULT_ROOM_PAGINATION_CREDITS, + room_pagination_batch_size: Self::DEFAULT_ROOM_PAGINATION_BATCH_SIZE, + experimental_auto_backpagination: false, } } } @@ -415,7 +468,7 @@ struct EventCacheInner { client: WeakClient, /// Global configuration for the event cache. - config: RwLock, + config: StdRwLock, /// Reference to the underlying store. store: EventCacheStoreLock, @@ -441,6 +494,9 @@ struct EventCacheInner { /// Needs to live here, so it may be passed to each [`RoomEventCache`] /// instance. /// + /// It's a `OnceLock` because its initialization is deferred to + /// [`EventCache::subscribe`]. + /// /// See doc comment of [`EventCache::auto_shrink_linked_chunk_task`]. auto_shrink_sender: OnceLock>, @@ -468,6 +524,12 @@ struct EventCacheInner { #[cfg(feature = "e2e-encryption")] redecryption_channels: redecryptor::RedecryptorChannels, + + /// State for the automatic pagination mechanism. + /// + /// Depends on the [`EventCacheConfig::experimental_auto_backpagination`] + /// flag to be set at subscription time. + automatic_pagination: OnceLock, } type AutoShrinkChannelPayload = OwnedRoomId; @@ -627,6 +689,7 @@ impl EventCacheInner { "we must have called `EventCache::subscribe()` before calling here.", ), self.store.clone(), + self.automatic_pagination.get().cloned(), ) .await?; diff --git a/crates/matrix-sdk/tests/integration/event_cache/mod.rs b/crates/matrix-sdk/tests/integration/event_cache/mod.rs index 7660d3cb9..0fa5e1bf5 100644 --- a/crates/matrix-sdk/tests/integration/event_cache/mod.rs +++ b/crates/matrix-sdk/tests/integration/event_cache/mod.rs @@ -602,7 +602,7 @@ async fn test_reset_while_backpaginating() { wait_for_initial_events(events, &mut room_stream).await; // We're going to cause a small race: - // - a background request to sync will be sent, + // - a pagination request to sync will be sent, // - a backpagination will be sent concurrently. // // So events have to happen in this order: diff --git a/crates/matrix-sdk/tests/integration/event_cache/read_receipts.rs b/crates/matrix-sdk/tests/integration/event_cache/read_receipts.rs index 34aa8796c..7c7c7482d 100644 --- a/crates/matrix-sdk/tests/integration/event_cache/read_receipts.rs +++ b/crates/matrix-sdk/tests/integration/event_cache/read_receipts.rs @@ -24,9 +24,12 @@ //! This avoids potential race conditions where a sync could be done, but the //! processing by the event cache isn't, at the time we check the unread counts. +use std::time::Duration; + use matrix_sdk::{ - ThreadingSupport, assert_let_timeout, event_cache::RoomEventCacheUpdate, - test_utils::mocks::MatrixMockServer, + ThreadingSupport, assert_let_timeout, + event_cache::RoomEventCacheUpdate, + test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate}, }; use matrix_sdk_test::{BOB, JoinedRoomBuilder, async_test, event_factory::EventFactory}; use ruma::{ @@ -715,3 +718,77 @@ async fn test_unread_counts_updated_after_duplicate_only_sync_response() { // The message counts are properly updated (zero new message unread after $2). assert_eq!(room.num_unread_messages(), 0); } + +/// Test that the unread count computation causes a back-pagination, when it +/// can't find an event pointed to by a read receipt. +#[async_test] +async fn test_compute_unread_counts_triggers_backpaginations() { + let server = MatrixMockServer::new().await; + let client = server.client_builder().build().await; + let own_user_id = client.user_id().unwrap(); + + client.event_cache().config_mut().experimental_auto_backpagination = true; + client.event_cache().subscribe().unwrap(); + + let room_id = room_id!("!omelette:fromage.fr"); + let f = EventFactory::new().room(room_id).sender(*BOB); + + let room = server.sync_joined_room(&client, room_id).await; + + let (room_event_cache, _drop_handles) = room.event_cache().await.unwrap(); + let (_, mut room_cache_updates) = room_event_cache.subscribe().await.unwrap(); + assert!(room_cache_updates.is_empty()); + + // Already set up the mock for /messages, as the background pagination will hit + // it as soon as the sync is received. + server + .mock_room_messages() + .match_from("prev_batch") + .ok(RoomMessagesResponseTemplate::default() + .events(vec![ + f.text_msg("hello 2").event_id(event_id!("$2")), + f.text_msg("hello 1").event_id(event_id!("$1")), + ]) + .with_delay(Duration::from_millis(100))) + .mock_once() + .mount() + .await; + + // Starting with a gappy sync, with a room with two messages from Bob, and a + // receipt on Bob's first message $1, which is missing from the timeline, + server + .sync_room( + &client, + JoinedRoomBuilder::new(room_id) + .add_timeline_event(f.text_msg("hello 3").event_id(event_id!("$3"))) + .add_timeline_event(f.text_msg("hello 4").event_id(event_id!("$4"))) + .add_receipt( + f.read_receipts() + .add( + event_id!("$1"), + own_user_id, + ReceiptType::Read, + ReceiptThread::Unthreaded, + ) + .into_event(), + ) + .set_timeline_limited() + .set_timeline_prev_batch("prev_batch".to_owned()), + ) + .await; + + // (one vector diff update, one read receipt update) + assert_let_timeout!(Ok(_) = room_cache_updates.recv()); + assert_let_timeout!(Ok(_) = room_cache_updates.recv()); + + // The message counts are properly updated (two unread messages in the + // timeline, which are $3 and $4). + assert_eq!(room.num_unread_messages(), 2); + + // Then, there's a background pagination happening in the room, which will fetch + // the missing $1 and $2. + assert_let_timeout!(Duration::from_millis(150), Ok(_) = room_cache_updates.recv()); + + // The message counts are properly updated (three messages after $1). + assert_eq!(room.num_unread_messages(), 3); +} diff --git a/labs/multiverse/src/main.rs b/labs/multiverse/src/main.rs index f5e315696..d20d2b417 100644 --- a/labs/multiverse/src/main.rs +++ b/labs/multiverse/src/main.rs @@ -153,6 +153,7 @@ async fn main() -> Result<()> { }); let event_cache = client.event_cache(); + event_cache.config_mut().experimental_auto_backpagination = true; event_cache.subscribe()?; let terminal = ratatui::init();