From 9f18d763551db43228dfa915336f054ba8690d74 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Wed, 18 Mar 2026 14:59:22 +0100 Subject: [PATCH 01/20] feat(event cache): add a feature toggle to enable automatic back-pagination --- bindings/matrix-sdk-ffi/src/client.rs | 9 +++++++++ crates/matrix-sdk/src/event_cache/mod.rs | 6 ++++++ 2 files changed, 15 insertions(+) diff --git a/bindings/matrix-sdk-ffi/src/client.rs b/bindings/matrix-sdk-ffi/src/client.rs index 71538d1ef..caf82cfe5 100644 --- a/bindings/matrix-sdk-ffi/src/client.rs +++ b/bindings/matrix-sdk-ffi/src/client.rs @@ -2104,6 +2104,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 async fn enable_automatic_backpagination(&self) { + self.inner.event_cache().config_mut().await.experimental_auto_backpagination = true; + } } #[cfg(feature = "experimental-element-recent-emojis")] diff --git a/crates/matrix-sdk/src/event_cache/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index 1cfcecb33..cb9a915d2 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -379,6 +379,11 @@ 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, } impl EventCacheConfig { @@ -395,6 +400,7 @@ 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, + experimental_auto_backpagination: false, } } } From bf9ffd7d09dca14e92af3529c78acd24b3c75fa3 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 23 Mar 2026 15:21:21 +0100 Subject: [PATCH 02/20] refactor(event cache): make `EventCache::config/config_mut` sync by using a sync lock There wasn't good reason to use an async lock, as this lock is always super short-lived, it can be sync, which avoids complications in the subsequent commit when calling sync init code. --- bindings/matrix-sdk-ffi/src/client.rs | 4 ++-- .../tests/integration/timeline/pinned_event.rs | 4 ++-- .../src/event_cache/caches/pinned_events/mod.rs | 2 +- crates/matrix-sdk/src/event_cache/mod.rs | 14 +++++++------- 4 files changed, 12 insertions(+), 12 deletions(-) diff --git a/bindings/matrix-sdk-ffi/src/client.rs b/bindings/matrix-sdk-ffi/src/client.rs index caf82cfe5..b3e7814da 100644 --- a/bindings/matrix-sdk-ffi/src/client.rs +++ b/bindings/matrix-sdk-ffi/src/client.rs @@ -2110,8 +2110,8 @@ impl Client { /// /// This is an experimental feature, and might cause performance issues on /// large accounts. Use with caution. - pub async fn enable_automatic_backpagination(&self) { - self.inner.event_cache().config_mut().await.experimental_auto_backpagination = true; + pub fn enable_automatic_backpagination(&self) { + self.inner.event_cache().config_mut().experimental_auto_backpagination = true; } } 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/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/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index cb9a915d2..a0b2595d8 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -31,7 +31,7 @@ use std::{ collections::HashMap, fmt, ops::{Deref, DerefMut}, - sync::{Arc, OnceLock}, + sync::{Arc, OnceLock, RwLock as StdRwLock}, }; use futures_util::future::try_join_all; @@ -228,7 +228,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(), @@ -249,13 +249,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) -> impl Deref + '_ { + 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) -> impl DerefMut + '_ { + self.inner.config.write().unwrap() } /// Subscribes to updates that a thread subscription has been sent. @@ -413,7 +413,7 @@ struct EventCacheInner { client: WeakClient, /// Global configuration for the event cache. - config: RwLock, + config: StdRwLock, /// Reference to the underlying store. store: EventCacheStoreLock, From ff8b27f99be881c40d65f7c90274bad21970b478 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 23 Mar 2026 15:24:26 +0100 Subject: [PATCH 03/20] feat(event cache): add an automatic background requests task --- .../matrix-sdk/src/event_cache/caches/mod.rs | 4 +- .../src/event_cache/caches/room/state.rs | 17 ++++- crates/matrix-sdk/src/event_cache/mod.rs | 35 +++++++++ crates/matrix-sdk/src/event_cache/tasks.rs | 73 ++++++++++++++++++- 4 files changed, 125 insertions(+), 4 deletions(-) diff --git a/crates/matrix-sdk/src/event_cache/caches/mod.rs b/crates/matrix-sdk/src/event_cache/caches/mod.rs index 90d494590..1517d88fd 100644 --- a/crates/matrix-sdk/src/event_cache/caches/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/mod.rs @@ -24,7 +24,7 @@ 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::tasks::BackgroundRequest, room::WeakRoom}; pub mod event_focused; pub mod event_linked_chunk; @@ -50,6 +50,7 @@ impl Caches { linked_chunk_update_sender: Sender, auto_shrink_sender: mpsc::Sender, store: EventCacheStoreLock, + background_request_sender: Option>, ) -> Result { let Some(client) = weak_client.get() else { return Err(EventCacheError::ClientDropped); @@ -84,6 +85,7 @@ impl Caches { linked_chunk_update_sender, store, pagination_status.clone(), + background_request_sender, ) .await?; 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 097f6b1ae..1593c5e44 100644 --- a/crates/matrix-sdk/src/event_cache/caches/room/state.rs +++ b/crates/matrix-sdk/src/event_cache/caches/room/state.rs @@ -49,7 +49,10 @@ use ruma::{ room_version_rules::RoomVersionRules, serde::Raw, }; -use tokio::sync::broadcast::{Receiver, Sender}; +use tokio::sync::{ + broadcast::{Receiver, Sender}, + mpsc, +}; use tracing::{debug, error, instrument, trace, warn}; use super::{ @@ -71,7 +74,11 @@ use super::{ RoomEventCacheLinkedChunkUpdate, RoomEventCacheUpdate, RoomEventCacheUpdateSender, sort_positions_descending, }; -use crate::{Room, event_cache::caches::pagination::SharedPaginationStatus, room::WeakRoom}; +use crate::{ + Room, + event_cache::{caches::pagination::SharedPaginationStatus, tasks::BackgroundRequest}, + room::WeakRoom, +}; /// Key for the event-focused caches. #[derive(Hash, PartialEq, Eq)] @@ -141,6 +148,10 @@ pub struct RoomEventCacheState { /// An atomic count of the current number of subscriber of the /// [`super::RoomEventCache`]. subscriber_count: Arc, + + /// A notifier to trigger backpagination under certain predefined + /// conditions. + background_request_sender: Option>, } impl RoomEventCacheState { @@ -301,6 +312,7 @@ impl LockedRoomEventCacheState { linked_chunk_update_sender: Sender, store: EventCacheStoreLock, pagination_status: SharedObservable, + background_request_sender: Option>, ) -> Result { let store_guard = match store.lock().await? { // Lock is clean: all good! @@ -383,6 +395,7 @@ impl LockedRoomEventCacheState { waited_for_initial_prev_token: false, subscriber_count: Default::default(), pinned_event_cache: OnceLock::new(), + background_request_sender, })) } } diff --git a/crates/matrix-sdk/src/event_cache/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index a0b2595d8..bc0ec793f 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -54,6 +54,7 @@ use tracing::{error, instrument, trace}; use crate::{ Client, client::{ClientInner, WeakClient}, + event_cache::tasks::BackgroundRequest, paginators::PaginatorError, }; @@ -153,6 +154,10 @@ pub struct EventCacheDropHandles { /// The task used to automatically shrink the linked chunks. auto_shrink_linked_chunk_task: BackgroundTaskHandle, + /// The task used to automatically handle background requests (like + /// paginations). + background_requests_task: Option, + /// The task used to automatically redecrypt UTDs. #[cfg(feature = "e2e-encryption")] _redecryptor: redecryptor::Redecryptor, @@ -169,6 +174,9 @@ impl Drop for EventCacheDropHandles { self.listen_updates_task.abort(); self.ignore_user_list_update_task.abort(); self.auto_shrink_linked_chunk_task.abort(); + if let Some(task) = self.background_requests_task.take() { + task.abort(); + } } } @@ -234,6 +242,7 @@ impl EventCache { by_room: Default::default(), drop_handles: Default::default(), auto_shrink_sender: Default::default(), + background_requests_sender: Default::default(), generic_update_sender, linked_chunk_update_sender, _thread_subscriber_task: thread_subscriber_task, @@ -312,6 +321,21 @@ impl EventCache { redecryptor::Redecryptor::new(&client, Arc::downgrade(&self.inner), receiver, &self.inner.linked_chunk_update_sender) }; + let background_requests_task = if self.config().experimental_auto_backpagination { + let (sender, receiver) = mpsc::channel(4096); + + // Run the deferred initialization of the background request sender, that is shared + // with every room. + self.inner.background_requests_sender.get_or_init(|| sender); + + trace!("spawning the backgrounds requests task"); + Some(task_monitor.spawn_background_task("event_cache::background_requests_task", tasks::background_requests_task( + self.inner.clone(), receiver + ))) + } else { + trace!("backgrounds requests task is disabled"); + None + }; Arc::new(EventCacheDropHandles { listen_updates_task, @@ -319,6 +343,7 @@ impl EventCache { auto_shrink_linked_chunk_task, #[cfg(feature = "e2e-encryption")] _redecryptor: redecryptor, + background_requests_task }) }); @@ -439,9 +464,18 @@ 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>, + /// A sender for background requests, that is shared with every room. + /// + /// It's a `OnceLock` because its initialization is deferred to + /// [`EventCache::subscribe`]. + background_requests_sender: OnceLock>, + /// A sender for room generic update. /// /// See doc comment of [`RoomEventCacheGenericUpdate`] and @@ -643,6 +677,7 @@ impl EventCacheInner { "we must have called `EventCache::subscribe()` before calling here.", ), self.store.clone(), + self.background_requests_sender.get().cloned(), ) .await?; diff --git a/crates/matrix-sdk/src/event_cache/tasks.rs b/crates/matrix-sdk/src/event_cache/tasks.rs index 95d6526fe..aecf8a669 100644 --- a/crates/matrix-sdk/src/event_cache/tasks.rs +++ b/crates/matrix-sdk/src/event_cache/tasks.rs @@ -22,7 +22,7 @@ use matrix_sdk_base::{ linked_chunk::OwnedLinkedChunkId, serde_helpers::extract_thread_root_from_content, sync::RoomUpdates, }; -use ruma::{OwnedEventId, OwnedTransactionId}; +use ruma::{OwnedEventId, OwnedRoomId, OwnedTransactionId}; use tokio::{ select, sync::{ @@ -88,6 +88,77 @@ pub(super) async fn room_updates_task( } } +#[derive(Clone, Debug)] +pub(crate) enum BackgroundRequest { + PaginateRoomBackwards { room_id: OwnedRoomId }, +} + +/// The maximum number of allowed room paginations, for a given room, that can +/// be executed in the background request 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). +const DEFAULT_ROOM_PAGINATION_CREDITS: usize = 20; + +/// The number of messages to paginate in a single batch, when executing a +/// background pagination request. +const DEFAULT_BACKGROUND_PAGINATION_SIZE: u16 = 30; + +/// Listen to background requests, and execute them in real-time. +#[instrument(skip_all)] +pub(super) async fn background_requests_task( + inner: Arc, + mut receiver: mpsc::Receiver, +) { + trace!("Spawning the background request task"); + + let mut room_pagination_credits = HashMap::new(); + + while let Some(request) = receiver.recv().await { + match request { + BackgroundRequest::PaginateRoomBackwards { room_id } => { + let credits = room_pagination_credits + .entry(room_id.clone()) + .or_insert(DEFAULT_ROOM_PAGINATION_CREDITS); + + 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(DEFAULT_BACKGROUND_PAGINATION_SIZE).await { + Ok(outcome) => { + // Background 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 background request task because receiver closed"); +} + /// Listen to _ignore user list update changes_ to clear the rooms when a user /// is ignored or unignored. #[instrument(skip_all)] From ac5552807b180f52cc685194c385a9f599744b79 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 23 Mar 2026 15:27:18 +0100 Subject: [PATCH 04/20] refactor(event cache): have `select_best_receipt` return a result struct This makes it simpler to test its expected behavior, and will make it trivial to add a system to request automatic paginations in the background. --- .../src/event_cache/caches/read_receipts.rs | 94 +++++++++++++------ 1 file changed, 65 insertions(+), 29 deletions(-) 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..e5add83a9 100644 --- a/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs +++ b/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs @@ -219,6 +219,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 +249,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 +277,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 +298,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 +315,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 +332,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 @@ -343,27 +366,24 @@ pub(crate) fn compute_unread_counts( ) { 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 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 +864,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 +903,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 +942,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 +986,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 +1031,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 +1074,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 +1126,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()); } } From c4a95f9bbea1e0519207b0fb5ff0716f35f0c825 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 23 Mar 2026 15:27:39 +0100 Subject: [PATCH 05/20] feat(event cache): automatically request paginations when no receipt has been found when computing unread counts --- .../src/event_cache/caches/read_receipts.rs | 16 +++++++++++++++- .../src/event_cache/caches/room/state.rs | 10 +++++----- 2 files changed, 20 insertions(+), 6 deletions(-) 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 e5add83a9..54ae84e1f 100644 --- a/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs +++ b/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs @@ -108,9 +108,10 @@ use ruma::{ }, serde::Raw, }; +use tokio::sync::mpsc::Sender; use tracing::{debug, instrument, trace, warn}; -use crate::event_cache::caches::event_linked_chunk::EventLinkedChunk; +use crate::event_cache::{caches::event_linked_chunk::EventLinkedChunk, tasks::BackgroundRequest}; trait RoomReadReceiptsExt { /// Update the [`RoomReadReceipts`] unread counts according to the new @@ -363,6 +364,7 @@ pub(crate) fn compute_unread_counts( linked_chunk: &EventLinkedChunk, read_receipts: &mut RoomReadReceipts, with_threading_support: bool, + background_request_sender: Option<&Sender>, ) { debug!(?read_receipts, "Starting"); @@ -375,6 +377,18 @@ pub(crate) fn compute_unread_counts( with_threading_support, ); + if select_best_receipt_result.request_pagination { + trace!("Requesting pagination to find a better receipt"); + // Note: we use `try_send` here to keep the method sync, as computing the + // perfect receipt is best effort. + if let Some(sender) = background_request_sender + && let Err(err) = sender + .try_send(BackgroundRequest::PaginateRoomBackwards { room_id: room_id.to_owned() }) + { + warn!(%err, "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 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 1593c5e44..c15c2d2c3 100644 --- a/crates/matrix-sdk/src/event_cache/caches/room/state.rs +++ b/crates/matrix-sdk/src/event_cache/caches/room/state.rs @@ -1046,9 +1046,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(); @@ -1057,9 +1056,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.background_request_sender.as_ref(), ); if prev_read_receipts != read_receipts { From 161750f6d0ec25d35b396f79817f01e6fefa2d7a Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 23 Mar 2026 16:52:31 +0100 Subject: [PATCH 06/20] refactor(event cache): add PartialEq to `EventsOrigin` --- crates/matrix-sdk/src/event_cache/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/matrix-sdk/src/event_cache/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index bc0ec793f..bcb7dcbe8 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -693,7 +693,7 @@ impl EventCacheInner { } /// Indicate where events are coming from. -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq)] pub enum EventsOrigin { /// Events are coming from a sync. Sync, From 856b2be99241d4f1d272b674f1212469679bbd69 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 23 Mar 2026 16:54:00 +0100 Subject: [PATCH 07/20] test(event cache): unit test automatic background pagination --- crates/matrix-sdk/src/event_cache/tasks.rs | 85 ++++++++++++++++++++++ 1 file changed, 85 insertions(+) diff --git a/crates/matrix-sdk/src/event_cache/tasks.rs b/crates/matrix-sdk/src/event_cache/tasks.rs index aecf8a669..bcfc592b1 100644 --- a/crates/matrix-sdk/src/event_cache/tasks.rs +++ b/crates/matrix-sdk/src/event_cache/tasks.rs @@ -597,3 +597,88 @@ pub(super) async fn search_indexing_task( } } } + +// MatrixMockServer et al. aren't available on wasm. +#[cfg(all(test, not(target_arch = "wasm32")))] +mod tests { + use matrix_sdk_test::{BOB, JoinedRoomBuilder, async_test, event_factory::EventFactory}; + use ruma::{event_id, room_id}; + use tokio::sync::mpsc; + + use crate::{ + assert_let_timeout, + event_cache::{ + EventsOrigin, RoomEventCacheUpdate, tasks::BackgroundRequest::PaginateRoomBackwards, + }, + test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate}, + }; + + impl super::super::EventCache { + fn background_requests_sender(&self) -> Option> { + self.inner.background_requests_sender.get().cloned() + } + } + + /// Test that we can send background requests and trigger room paginations. + #[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_room( + &client, + JoinedRoomBuilder::new(room_id) + .set_timeline_limited() + .set_timeline_prev_batch("prev_batch"), + ) + .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()); + + // 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 sender = event_cache.background_requests_sender().unwrap(); + sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).await.unwrap(); + + assert_let_timeout!( + Ok(RoomEventCacheUpdate::UpdateTimelineEvents(update)) = room_cache_updates.recv() + ); + assert_eq!(update.diffs.len(), 1); + + assert_eq!(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")); + + assert!(room_cache_updates.is_empty()); + } +} From bd7dca45ef3aa8fed0bda2c23b6be6e971468bda Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 23 Mar 2026 17:07:37 +0100 Subject: [PATCH 08/20] feat(event cache): make some background pagination parameters configurable --- crates/matrix-sdk/src/event_cache/mod.rs | 32 ++++++++++++++++++++-- crates/matrix-sdk/src/event_cache/tasks.rs | 18 +++--------- 2 files changed, 34 insertions(+), 16 deletions(-) diff --git a/crates/matrix-sdk/src/event_cache/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index bcb7dcbe8..a819fed97 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -409,15 +409,41 @@ pub struct EventCacheConfig { /// /// 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 background request 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 a + /// background 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 background + /// 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 a background pagination request (see also + /// [`EventCacheConfig::room_pagination_batch_size`]). + pub const DEFAULT_ROOM_PAGINATION_BATCH_SIZE: u16 = 30; } impl Default for EventCacheConfig { @@ -425,6 +451,8 @@ 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, } } diff --git a/crates/matrix-sdk/src/event_cache/tasks.rs b/crates/matrix-sdk/src/event_cache/tasks.rs index bcfc592b1..c6fe582f6 100644 --- a/crates/matrix-sdk/src/event_cache/tasks.rs +++ b/crates/matrix-sdk/src/event_cache/tasks.rs @@ -93,18 +93,6 @@ pub(crate) enum BackgroundRequest { PaginateRoomBackwards { room_id: OwnedRoomId }, } -/// The maximum number of allowed room paginations, for a given room, that can -/// be executed in the background request 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). -const DEFAULT_ROOM_PAGINATION_CREDITS: usize = 20; - -/// The number of messages to paginate in a single batch, when executing a -/// background pagination request. -const DEFAULT_BACKGROUND_PAGINATION_SIZE: u16 = 30; - /// Listen to background requests, and execute them in real-time. #[instrument(skip_all)] pub(super) async fn background_requests_task( @@ -118,9 +106,11 @@ pub(super) async fn background_requests_task( while let Some(request) = receiver.recv().await { match request { BackgroundRequest::PaginateRoomBackwards { room_id } => { + let config = *inner.config.read().unwrap(); + let credits = room_pagination_credits .entry(room_id.clone()) - .or_insert(DEFAULT_ROOM_PAGINATION_CREDITS); + .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"); @@ -137,7 +127,7 @@ pub(super) async fn background_requests_task( trace!(for_room = %room_id, "automatic backpagination triggered"); - match pagination.run_backwards_once(DEFAULT_BACKGROUND_PAGINATION_SIZE).await { + match pagination.run_backwards_once(config.room_pagination_batch_size).await { Ok(outcome) => { // Background requests must be idempotent, so we only decrement credits if // we actually paginated something new. From d1d174d1379e9927840f5ef9671bcb054bec3067 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 23 Mar 2026 17:07:45 +0100 Subject: [PATCH 09/20] test(event cache): test the credit system of automatic pagination --- crates/matrix-sdk/src/event_cache/tasks.rs | 102 +++++++++++++++++++++ 1 file changed, 102 insertions(+) diff --git a/crates/matrix-sdk/src/event_cache/tasks.rs b/crates/matrix-sdk/src/event_cache/tasks.rs index c6fe582f6..a2bcb2124 100644 --- a/crates/matrix-sdk/src/event_cache/tasks.rs +++ b/crates/matrix-sdk/src/event_cache/tasks.rs @@ -591,6 +591,9 @@ pub(super) async fn search_indexing_task( // MatrixMockServer et al. aren't available on wasm. #[cfg(all(test, not(target_arch = "wasm32")))] mod tests { + use std::time::Duration; + + use matrix_sdk_base::sleep::sleep; use matrix_sdk_test::{BOB, JoinedRoomBuilder, async_test, event_factory::EventFactory}; use ruma::{event_id, room_id}; use tokio::sync::mpsc; @@ -653,6 +656,7 @@ mod tests { let sender = event_cache.background_requests_sender().unwrap(); sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).await.unwrap(); + // The room pagination happens in the background. assert_let_timeout!( Ok(RoomEventCacheUpdate::UpdateTimelineEvents(update)) = room_cache_updates.recv() ); @@ -669,6 +673,104 @@ mod tests { 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(_)) = room_cache_updates.recv() + ); + + // 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 sender = event_cache.background_requests_sender().unwrap(); + sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).await.unwrap(); + + // 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_eq!(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… + sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).await.unwrap(); + + 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); + } } From 871cb2221b3e151c73116a9b1749072dcf375ca6 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 23 Mar 2026 17:19:20 +0100 Subject: [PATCH 10/20] test(event cache): add an integration test for automatically paginating a room with missing receipts --- .../integration/event_cache/read_receipts.rs | 81 ++++++++++++++++++- 1 file changed, 79 insertions(+), 2 deletions(-) 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 8f25bf019..74a2f9381 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::{ @@ -652,3 +655,77 @@ async fn test_select_best_receipt_considers_thread_config() { // implicit receipt sent in a thread isn't taken into account. assert_eq!(room.num_unread_messages(), 2); } + +/// 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); +} From 87a2f8fab7e73a85d3c9bc0d0898b22c2ca41507 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 23 Mar 2026 17:37:10 +0100 Subject: [PATCH 11/20] feat(multiverse): enable automatic back pagination in multiverse --- labs/multiverse/src/main.rs | 1 + 1 file changed, 1 insertion(+) 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(); From 4feeaf0cf5aec2d9ab326498adb0a60f67f4e376 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 30 Mar 2026 11:01:57 +0200 Subject: [PATCH 12/20] chore(review): address first review comments --- crates/matrix-sdk/src/event_cache/mod.rs | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/crates/matrix-sdk/src/event_cache/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index a819fed97..48780dcf8 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, RwLock as StdRwLock}, + sync::{Arc, OnceLock, RwLock as StdRwLock, RwLockReadGuard, RwLockWriteGuard}, }; use futures_util::future::try_join_all; @@ -258,12 +257,12 @@ impl EventCache { /// Get a read-only handle to the global configuration of the /// [`EventCache`]. - pub fn config(&self) -> impl Deref + '_ { + pub fn config(&self) -> RwLockReadGuard<'_, EventCacheConfig> { self.inner.config.read().unwrap() } /// Get a writable handle to the global configuration of the [`EventCache`]. - pub fn config_mut(&self) -> impl DerefMut + '_ { + pub fn config_mut(&self) -> RwLockWriteGuard<'_, EventCacheConfig> { self.inner.config.write().unwrap() } From 27415dc64d44f528b9df1bb55131e9fcdaf83502 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 30 Mar 2026 11:04:41 +0200 Subject: [PATCH 13/20] chore(review): Use assert_matches and remove PartialEq/Eq impls for `EventsOrigin` --- crates/matrix-sdk/src/event_cache/mod.rs | 2 +- crates/matrix-sdk/src/event_cache/tasks.rs | 5 +++-- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/crates/matrix-sdk/src/event_cache/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index 48780dcf8..b5f35b0a5 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -720,7 +720,7 @@ impl EventCacheInner { } /// Indicate where events are coming from. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone)] pub enum EventsOrigin { /// Events are coming from a sync. Sync, diff --git a/crates/matrix-sdk/src/event_cache/tasks.rs b/crates/matrix-sdk/src/event_cache/tasks.rs index a2bcb2124..4faa0ac67 100644 --- a/crates/matrix-sdk/src/event_cache/tasks.rs +++ b/crates/matrix-sdk/src/event_cache/tasks.rs @@ -593,6 +593,7 @@ pub(super) async fn search_indexing_task( mod tests { use std::time::Duration; + use assert_matches::assert_matches; use matrix_sdk_base::sleep::sleep; use matrix_sdk_test::{BOB, JoinedRoomBuilder, async_test, event_factory::EventFactory}; use ruma::{event_id, room_id}; @@ -662,7 +663,7 @@ mod tests { ); assert_eq!(update.diffs.len(), 1); - assert_eq!(update.origin, EventsOrigin::Pagination); + assert_matches!(update.origin, EventsOrigin::Pagination); let mut room_events = room_events.into(); for diff in update.diffs { @@ -739,7 +740,7 @@ mod tests { ); assert_eq!(update.diffs.len(), 1); - assert_eq!(update.origin, EventsOrigin::Pagination); + assert_matches!(update.origin, EventsOrigin::Pagination); let mut room_events = room_events.into(); for diff in update.diffs { From d151340882a94165f9c9ad45f94e168570106091 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 30 Mar 2026 11:12:39 +0200 Subject: [PATCH 14/20] chore(review): use an unbounded sender for `BackgroundRequest` --- crates/matrix-sdk/src/event_cache/caches/mod.rs | 2 +- .../src/event_cache/caches/read_receipts.rs | 11 ++++++----- .../matrix-sdk/src/event_cache/caches/room/state.rs | 4 ++-- crates/matrix-sdk/src/event_cache/mod.rs | 4 ++-- crates/matrix-sdk/src/event_cache/tasks.rs | 12 +++++++----- 5 files changed, 18 insertions(+), 15 deletions(-) diff --git a/crates/matrix-sdk/src/event_cache/caches/mod.rs b/crates/matrix-sdk/src/event_cache/caches/mod.rs index 1517d88fd..8679a2dd8 100644 --- a/crates/matrix-sdk/src/event_cache/caches/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/mod.rs @@ -50,7 +50,7 @@ impl Caches { linked_chunk_update_sender: Sender, auto_shrink_sender: mpsc::Sender, store: EventCacheStoreLock, - background_request_sender: Option>, + background_request_sender: Option>, ) -> Result { let Some(client) = weak_client.get() else { return Err(EventCacheError::ClientDropped); 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 54ae84e1f..a8839ecd4 100644 --- a/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs +++ b/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs @@ -108,7 +108,7 @@ use ruma::{ }, serde::Raw, }; -use tokio::sync::mpsc::Sender; +use tokio::sync::mpsc::UnboundedSender; use tracing::{debug, instrument, trace, warn}; use crate::event_cache::{caches::event_linked_chunk::EventLinkedChunk, tasks::BackgroundRequest}; @@ -364,7 +364,7 @@ pub(crate) fn compute_unread_counts( linked_chunk: &EventLinkedChunk, read_receipts: &mut RoomReadReceipts, with_threading_support: bool, - background_request_sender: Option<&Sender>, + background_request_sender: Option<&UnboundedSender>, ) { debug!(?read_receipts, "Starting"); @@ -382,10 +382,11 @@ pub(crate) fn compute_unread_counts( // Note: we use `try_send` here to keep the method sync, as computing the // perfect receipt is best effort. if let Some(sender) = background_request_sender - && let Err(err) = sender - .try_send(BackgroundRequest::PaginateRoomBackwards { room_id: room_id.to_owned() }) + && sender + .send(BackgroundRequest::PaginateRoomBackwards { room_id: room_id.to_owned() }) + .is_err() { - warn!(%err, "Failed to request pagination to find a better receipt"); + warn!("Failed to request pagination to find a better receipt"); } } 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 c15c2d2c3..a5f08a306 100644 --- a/crates/matrix-sdk/src/event_cache/caches/room/state.rs +++ b/crates/matrix-sdk/src/event_cache/caches/room/state.rs @@ -151,7 +151,7 @@ pub struct RoomEventCacheState { /// A notifier to trigger backpagination under certain predefined /// conditions. - background_request_sender: Option>, + background_request_sender: Option>, } impl RoomEventCacheState { @@ -312,7 +312,7 @@ impl LockedRoomEventCacheState { linked_chunk_update_sender: Sender, store: EventCacheStoreLock, pagination_status: SharedObservable, - background_request_sender: Option>, + background_request_sender: Option>, ) -> Result { let store_guard = match store.lock().await? { // Lock is clean: all good! diff --git a/crates/matrix-sdk/src/event_cache/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index b5f35b0a5..2c105c201 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -321,7 +321,7 @@ impl EventCache { }; let background_requests_task = if self.config().experimental_auto_backpagination { - let (sender, receiver) = mpsc::channel(4096); + let (sender, receiver) = mpsc::unbounded_channel(); // Run the deferred initialization of the background request sender, that is shared // with every room. @@ -501,7 +501,7 @@ struct EventCacheInner { /// /// It's a `OnceLock` because its initialization is deferred to /// [`EventCache::subscribe`]. - background_requests_sender: OnceLock>, + background_requests_sender: OnceLock>, /// A sender for room generic update. /// diff --git a/crates/matrix-sdk/src/event_cache/tasks.rs b/crates/matrix-sdk/src/event_cache/tasks.rs index 4faa0ac67..5e06c841b 100644 --- a/crates/matrix-sdk/src/event_cache/tasks.rs +++ b/crates/matrix-sdk/src/event_cache/tasks.rs @@ -97,7 +97,7 @@ pub(crate) enum BackgroundRequest { #[instrument(skip_all)] pub(super) async fn background_requests_task( inner: Arc, - mut receiver: mpsc::Receiver, + mut receiver: mpsc::UnboundedReceiver, ) { trace!("Spawning the background request task"); @@ -608,7 +608,9 @@ mod tests { }; impl super::super::EventCache { - fn background_requests_sender(&self) -> Option> { + fn background_requests_sender( + &self, + ) -> Option> { self.inner.background_requests_sender.get().cloned() } } @@ -655,7 +657,7 @@ mod tests { // Send a request for a background pagination, let sender = event_cache.background_requests_sender().unwrap(); - sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).await.unwrap(); + sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); // The room pagination happens in the background. assert_let_timeout!( @@ -732,7 +734,7 @@ mod tests { // Send a request for a background pagination, let sender = event_cache.background_requests_sender().unwrap(); - sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).await.unwrap(); + sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); // The room pagination happens in the background. assert_let_timeout!( @@ -755,7 +757,7 @@ mod tests { assert!(room_cache_updates.is_empty()); // One can send another request to back-paginate… - sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).await.unwrap(); + sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); sleep(Duration::from_millis(300)).await; // But it doesn't happen, because we don't have enough credits for automatic From 552639763eb85e55ba941a67caf6d1a9fdbef35f Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 30 Mar 2026 11:19:01 +0200 Subject: [PATCH 15/20] chore: move the `BackgroundRequest` type, tests and task to its own file --- .../src/event_cache/automatic_pagination.rs | 273 ++++++++++++++++++ .../matrix-sdk/src/event_cache/caches/mod.rs | 4 +- .../src/event_cache/caches/read_receipts.rs | 4 +- .../src/event_cache/caches/room/state.rs | 4 +- crates/matrix-sdk/src/event_cache/mod.rs | 5 +- crates/matrix-sdk/src/event_cache/tasks.rs | 253 +--------------- 6 files changed, 286 insertions(+), 257 deletions(-) create mode 100644 crates/matrix-sdk/src/event_cache/automatic_pagination.rs 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..51d7c482c --- /dev/null +++ b/crates/matrix-sdk/src/event_cache/automatic_pagination.rs @@ -0,0 +1,273 @@ +// 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}; + +use ruma::OwnedRoomId; +use tokio::sync::mpsc; +use tracing::{info, instrument, trace, warn}; + +use crate::event_cache::EventCacheInner; + +#[derive(Clone, Debug)] +pub(crate) enum BackgroundRequest { + PaginateRoomBackwards { room_id: OwnedRoomId }, +} + +/// Listen to background requests, and execute them in real-time. +#[instrument(skip_all)] +pub(super) async fn background_requests_task( + inner: Arc, + mut receiver: mpsc::UnboundedReceiver, +) { + trace!("Spawning the background request task"); + + let mut room_pagination_credits = HashMap::new(); + + while let Some(request) = receiver.recv().await { + match request { + BackgroundRequest::PaginateRoomBackwards { room_id } => { + 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) => { + // Background 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 background request 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 matrix_sdk_base::sleep::sleep; + use matrix_sdk_test::{BOB, JoinedRoomBuilder, async_test, event_factory::EventFactory}; + use ruma::{event_id, room_id}; + use tokio::sync::mpsc; + + use crate::{ + assert_let_timeout, + event_cache::{ + EventsOrigin, RoomEventCacheUpdate, + automatic_pagination::BackgroundRequest::PaginateRoomBackwards, + }, + test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate}, + }; + + impl super::super::EventCache { + fn background_requests_sender( + &self, + ) -> Option> { + self.inner.background_requests_sender.get().cloned() + } + } + + /// Test that we can send background requests and trigger room paginations. + #[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_room( + &client, + JoinedRoomBuilder::new(room_id) + .set_timeline_limited() + .set_timeline_prev_batch("prev_batch"), + ) + .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()); + + // 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 sender = event_cache.background_requests_sender().unwrap(); + sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); + + // 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(_)) = room_cache_updates.recv() + ); + + // 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 sender = event_cache.background_requests_sender().unwrap(); + sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); + + // 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… + sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); + + 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 8679a2dd8..c642feac6 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, event_cache::tasks::BackgroundRequest, room::WeakRoom}; +use crate::{ + client::WeakClient, event_cache::automatic_pagination::BackgroundRequest, room::WeakRoom, +}; pub mod event_focused; pub mod event_linked_chunk; 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 a8839ecd4..5b3806db9 100644 --- a/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs +++ b/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs @@ -111,7 +111,9 @@ use ruma::{ use tokio::sync::mpsc::UnboundedSender; use tracing::{debug, instrument, trace, warn}; -use crate::event_cache::{caches::event_linked_chunk::EventLinkedChunk, tasks::BackgroundRequest}; +use crate::event_cache::{ + automatic_pagination::BackgroundRequest, caches::event_linked_chunk::EventLinkedChunk, +}; trait RoomReadReceiptsExt { /// Update the [`RoomReadReceipts`] unread counts according to the new 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 a5f08a306..39dffe183 100644 --- a/crates/matrix-sdk/src/event_cache/caches/room/state.rs +++ b/crates/matrix-sdk/src/event_cache/caches/room/state.rs @@ -76,7 +76,9 @@ use super::{ }; use crate::{ Room, - event_cache::{caches::pagination::SharedPaginationStatus, tasks::BackgroundRequest}, + event_cache::{ + automatic_pagination::BackgroundRequest, caches::pagination::SharedPaginationStatus, + }, room::WeakRoom, }; diff --git a/crates/matrix-sdk/src/event_cache/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index 2c105c201..2dbd80850 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -53,10 +53,11 @@ use tracing::{error, instrument, trace}; use crate::{ Client, client::{ClientInner, WeakClient}, - event_cache::tasks::BackgroundRequest, + event_cache::automatic_pagination::{BackgroundRequest, background_requests_task}, paginators::PaginatorError, }; +mod automatic_pagination; mod caches; mod deduplicator; mod persistence; @@ -328,7 +329,7 @@ impl EventCache { self.inner.background_requests_sender.get_or_init(|| sender); trace!("spawning the backgrounds requests task"); - Some(task_monitor.spawn_background_task("event_cache::background_requests_task", tasks::background_requests_task( + Some(task_monitor.spawn_background_task("event_cache::background_requests_task", background_requests_task( self.inner.clone(), receiver ))) } else { diff --git a/crates/matrix-sdk/src/event_cache/tasks.rs b/crates/matrix-sdk/src/event_cache/tasks.rs index 5e06c841b..95d6526fe 100644 --- a/crates/matrix-sdk/src/event_cache/tasks.rs +++ b/crates/matrix-sdk/src/event_cache/tasks.rs @@ -22,7 +22,7 @@ use matrix_sdk_base::{ linked_chunk::OwnedLinkedChunkId, serde_helpers::extract_thread_root_from_content, sync::RoomUpdates, }; -use ruma::{OwnedEventId, OwnedRoomId, OwnedTransactionId}; +use ruma::{OwnedEventId, OwnedTransactionId}; use tokio::{ select, sync::{ @@ -88,67 +88,6 @@ pub(super) async fn room_updates_task( } } -#[derive(Clone, Debug)] -pub(crate) enum BackgroundRequest { - PaginateRoomBackwards { room_id: OwnedRoomId }, -} - -/// Listen to background requests, and execute them in real-time. -#[instrument(skip_all)] -pub(super) async fn background_requests_task( - inner: Arc, - mut receiver: mpsc::UnboundedReceiver, -) { - trace!("Spawning the background request task"); - - let mut room_pagination_credits = HashMap::new(); - - while let Some(request) = receiver.recv().await { - match request { - BackgroundRequest::PaginateRoomBackwards { room_id } => { - 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) => { - // Background 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 background request task because receiver closed"); -} - /// Listen to _ignore user list update changes_ to clear the rooms when a user /// is ignored or unignored. #[instrument(skip_all)] @@ -587,193 +526,3 @@ pub(super) async fn search_indexing_task( } } } - -// 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 matrix_sdk_base::sleep::sleep; - use matrix_sdk_test::{BOB, JoinedRoomBuilder, async_test, event_factory::EventFactory}; - use ruma::{event_id, room_id}; - use tokio::sync::mpsc; - - use crate::{ - assert_let_timeout, - event_cache::{ - EventsOrigin, RoomEventCacheUpdate, tasks::BackgroundRequest::PaginateRoomBackwards, - }, - test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate}, - }; - - impl super::super::EventCache { - fn background_requests_sender( - &self, - ) -> Option> { - self.inner.background_requests_sender.get().cloned() - } - } - - /// Test that we can send background requests and trigger room paginations. - #[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_room( - &client, - JoinedRoomBuilder::new(room_id) - .set_timeline_limited() - .set_timeline_prev_batch("prev_batch"), - ) - .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()); - - // 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 sender = event_cache.background_requests_sender().unwrap(); - sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); - - // 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(_)) = room_cache_updates.recv() - ); - - // 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 sender = event_cache.background_requests_sender().unwrap(); - sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); - - // 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… - sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); - - 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); - } -} From 752695c6ff9c61334e31931a5030ec45ffaba635 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 30 Mar 2026 11:27:28 +0200 Subject: [PATCH 16/20] chore: rename background request to automatic pagination request And associated vocabulary and fields. --- .../src/event_cache/automatic_pagination.rs | 32 ++++++------- .../matrix-sdk/src/event_cache/caches/mod.rs | 9 ++-- .../src/event_cache/caches/read_receipts.rs | 12 ++--- .../src/event_cache/caches/room/state.rs | 17 ++++--- crates/matrix-sdk/src/event_cache/mod.rs | 45 ++++++++++--------- .../tests/integration/event_cache/mod.rs | 2 +- 6 files changed, 63 insertions(+), 54 deletions(-) diff --git a/crates/matrix-sdk/src/event_cache/automatic_pagination.rs b/crates/matrix-sdk/src/event_cache/automatic_pagination.rs index 51d7c482c..931dd65b8 100644 --- a/crates/matrix-sdk/src/event_cache/automatic_pagination.rs +++ b/crates/matrix-sdk/src/event_cache/automatic_pagination.rs @@ -21,23 +21,24 @@ use tracing::{info, instrument, trace, warn}; use crate::event_cache::EventCacheInner; #[derive(Clone, Debug)] -pub(crate) enum BackgroundRequest { +pub(crate) enum AutomaticPaginationRequest { PaginateRoomBackwards { room_id: OwnedRoomId }, } -/// Listen to background requests, and execute them in real-time. +/// Listen to background automatic pagination requests, and execute them in +/// real-time. #[instrument(skip_all)] -pub(super) async fn background_requests_task( +pub(super) async fn automatic_paginations_task( inner: Arc, - mut receiver: mpsc::UnboundedReceiver, + mut receiver: mpsc::UnboundedReceiver, ) { - trace!("Spawning the background request task"); + trace!("Spawning the automatic pagination task"); let mut room_pagination_credits = HashMap::new(); while let Some(request) = receiver.recv().await { match request { - BackgroundRequest::PaginateRoomBackwards { room_id } => { + AutomaticPaginationRequest::PaginateRoomBackwards { room_id } => { let config = *inner.config.read().unwrap(); let credits = room_pagination_credits @@ -61,12 +62,13 @@ pub(super) async fn background_requests_task( match pagination.run_backwards_once(config.room_pagination_batch_size).await { Ok(outcome) => { - // Background requests must be idempotent, so we only decrement credits if + // 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 @@ -78,7 +80,7 @@ pub(super) async fn background_requests_task( } // The sender has shut down, exit. - info!("Closing the background request task because receiver closed"); + info!("Closing the automatic pagination task because receiver closed"); } // MatrixMockServer et al. aren't available on wasm. @@ -96,20 +98,20 @@ mod tests { assert_let_timeout, event_cache::{ EventsOrigin, RoomEventCacheUpdate, - automatic_pagination::BackgroundRequest::PaginateRoomBackwards, + automatic_pagination::AutomaticPaginationRequest::PaginateRoomBackwards, }, test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate}, }; impl super::super::EventCache { - fn background_requests_sender( + fn pagination_requests_sender( &self, - ) -> Option> { - self.inner.background_requests_sender.get().cloned() + ) -> Option> { + self.inner.automatic_pagination_requests_sender.get().cloned() } } - /// Test that we can send background requests and trigger room paginations. + /// Test that we can send automatic pagination requests. #[async_test] async fn test_background_room_paginations() { let server = MatrixMockServer::new().await; @@ -150,7 +152,7 @@ mod tests { .await; // Send a request for a background pagination, - let sender = event_cache.background_requests_sender().unwrap(); + let sender = event_cache.pagination_requests_sender().unwrap(); sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); // The room pagination happens in the background. @@ -227,7 +229,7 @@ mod tests { .await; // Send a request for a background pagination, - let sender = event_cache.background_requests_sender().unwrap(); + let sender = event_cache.pagination_requests_sender().unwrap(); sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); // The room pagination happens in the background. diff --git a/crates/matrix-sdk/src/event_cache/caches/mod.rs b/crates/matrix-sdk/src/event_cache/caches/mod.rs index c642feac6..56c5e9a97 100644 --- a/crates/matrix-sdk/src/event_cache/caches/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/mod.rs @@ -25,7 +25,8 @@ use tokio::sync::{broadcast::Sender, mpsc}; use super::{EventCacheError, EventsOrigin, Result}; use crate::{ - client::WeakClient, event_cache::automatic_pagination::BackgroundRequest, room::WeakRoom, + client::WeakClient, event_cache::automatic_pagination::AutomaticPaginationRequest, + room::WeakRoom, }; pub mod event_focused; @@ -52,7 +53,9 @@ impl Caches { linked_chunk_update_sender: Sender, auto_shrink_sender: mpsc::Sender, store: EventCacheStoreLock, - background_request_sender: Option>, + automatic_pagination_requests_sender: Option< + mpsc::UnboundedSender, + >, ) -> Result { let Some(client) = weak_client.get() else { return Err(EventCacheError::ClientDropped); @@ -87,7 +90,7 @@ impl Caches { linked_chunk_update_sender, store, pagination_status.clone(), - background_request_sender, + automatic_pagination_requests_sender, ) .await?; 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 5b3806db9..f839a8118 100644 --- a/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs +++ b/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs @@ -112,7 +112,7 @@ use tokio::sync::mpsc::UnboundedSender; use tracing::{debug, instrument, trace, warn}; use crate::event_cache::{ - automatic_pagination::BackgroundRequest, caches::event_linked_chunk::EventLinkedChunk, + automatic_pagination::AutomaticPaginationRequest, caches::event_linked_chunk::EventLinkedChunk, }; trait RoomReadReceiptsExt { @@ -366,7 +366,7 @@ pub(crate) fn compute_unread_counts( linked_chunk: &EventLinkedChunk, read_receipts: &mut RoomReadReceipts, with_threading_support: bool, - background_request_sender: Option<&UnboundedSender>, + automatic_pagination_request_sender: Option<&UnboundedSender>, ) { debug!(?read_receipts, "Starting"); @@ -381,11 +381,11 @@ pub(crate) fn compute_unread_counts( if select_best_receipt_result.request_pagination { trace!("Requesting pagination to find a better receipt"); - // Note: we use `try_send` here to keep the method sync, as computing the - // perfect receipt is best effort. - if let Some(sender) = background_request_sender + if let Some(sender) = automatic_pagination_request_sender && sender - .send(BackgroundRequest::PaginateRoomBackwards { room_id: room_id.to_owned() }) + .send(AutomaticPaginationRequest::PaginateRoomBackwards { + room_id: room_id.to_owned(), + }) .is_err() { warn!("Failed to request pagination to find a better receipt"); 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 39dffe183..8678c92f5 100644 --- a/crates/matrix-sdk/src/event_cache/caches/room/state.rs +++ b/crates/matrix-sdk/src/event_cache/caches/room/state.rs @@ -77,7 +77,8 @@ use super::{ use crate::{ Room, event_cache::{ - automatic_pagination::BackgroundRequest, caches::pagination::SharedPaginationStatus, + automatic_pagination::AutomaticPaginationRequest, + caches::pagination::SharedPaginationStatus, }, room::WeakRoom, }; @@ -151,9 +152,9 @@ pub struct RoomEventCacheState { /// [`super::RoomEventCache`]. subscriber_count: Arc, - /// A notifier to trigger backpagination under certain predefined - /// conditions. - background_request_sender: Option>, + /// A sender to trigger automatic pagination requests under certain + /// predefined conditions. + automatic_pagination_request_sender: Option>, } impl RoomEventCacheState { @@ -314,7 +315,9 @@ impl LockedRoomEventCacheState { linked_chunk_update_sender: Sender, store: EventCacheStoreLock, pagination_status: SharedObservable, - background_request_sender: Option>, + automatic_pagination_request_sender: Option< + mpsc::UnboundedSender, + >, ) -> Result { let store_guard = match store.lock().await? { // Lock is clean: all good! @@ -397,7 +400,7 @@ impl LockedRoomEventCacheState { waited_for_initial_prev_token: false, subscriber_count: Default::default(), pinned_event_cache: OnceLock::new(), - background_request_sender, + automatic_pagination_request_sender, })) } } @@ -1061,7 +1064,7 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> { &self.state.room_linked_chunk, &mut read_receipts, self.state.enabled_thread_support, - self.state.background_request_sender.as_ref(), + self.state.automatic_pagination_request_sender.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 2dbd80850..7d97d639f 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -53,7 +53,7 @@ use tracing::{error, instrument, trace}; use crate::{ Client, client::{ClientInner, WeakClient}, - event_cache::automatic_pagination::{BackgroundRequest, background_requests_task}, + event_cache::automatic_pagination::{AutomaticPaginationRequest, automatic_paginations_task}, paginators::PaginatorError, }; @@ -154,9 +154,8 @@ pub struct EventCacheDropHandles { /// The task used to automatically shrink the linked chunks. auto_shrink_linked_chunk_task: BackgroundTaskHandle, - /// The task used to automatically handle background requests (like - /// paginations). - background_requests_task: Option, + /// The task used to handle automatic pagination requests. + automatic_paginations_task: Option, /// The task used to automatically redecrypt UTDs. #[cfg(feature = "e2e-encryption")] @@ -174,7 +173,7 @@ impl Drop for EventCacheDropHandles { self.listen_updates_task.abort(); self.ignore_user_list_update_task.abort(); self.auto_shrink_linked_chunk_task.abort(); - if let Some(task) = self.background_requests_task.take() { + if let Some(task) = self.automatic_paginations_task.take() { task.abort(); } } @@ -242,7 +241,7 @@ impl EventCache { by_room: Default::default(), drop_handles: Default::default(), auto_shrink_sender: Default::default(), - background_requests_sender: Default::default(), + automatic_pagination_requests_sender: Default::default(), generic_update_sender, linked_chunk_update_sender, _thread_subscriber_task: thread_subscriber_task, @@ -321,19 +320,19 @@ impl EventCache { redecryptor::Redecryptor::new(&client, Arc::downgrade(&self.inner), receiver, &self.inner.linked_chunk_update_sender) }; - let background_requests_task = if self.config().experimental_auto_backpagination { + let automatic_paginations_task = if self.config().experimental_auto_backpagination { let (sender, receiver) = mpsc::unbounded_channel(); - // Run the deferred initialization of the background request sender, that is shared - // with every room. - self.inner.background_requests_sender.get_or_init(|| sender); + // Run the deferred initialization of the automatic pagination request sender, that + // is shared with every room. + self.inner.automatic_pagination_requests_sender.get_or_init(|| sender); - trace!("spawning the backgrounds requests task"); - Some(task_monitor.spawn_background_task("event_cache::background_requests_task", background_requests_task( + trace!("spawning the automatic paginations task"); + Some(task_monitor.spawn_background_task("event_cache::automatic_paginations_task", automatic_paginations_task( self.inner.clone(), receiver ))) } else { - trace!("backgrounds requests task is disabled"); + trace!("automatic paginations task is disabled"); None }; @@ -343,7 +342,7 @@ impl EventCache { auto_shrink_linked_chunk_task, #[cfg(feature = "e2e-encryption")] _redecryptor: redecryptor, - background_requests_task + automatic_paginations_task }) }); @@ -411,7 +410,7 @@ pub struct EventCacheConfig { pub experimental_auto_backpagination: bool, /// The maximum number of allowed room paginations, for a given room, that - /// can be executed in the background request task. + /// 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 @@ -420,8 +419,8 @@ pub struct EventCacheConfig { /// 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 a - /// background pagination request. + /// 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, @@ -435,13 +434,13 @@ impl EventCacheConfig { /// loading the pinned events. pub const DEFAULT_MAX_CONCURRENT_REQUESTS: usize = 8; - /// The default number of credits to give to a room for background + /// 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 a background pagination request (see also + /// executing an automatic pagination request (see also /// [`EventCacheConfig::room_pagination_batch_size`]). pub const DEFAULT_ROOM_PAGINATION_BATCH_SIZE: u16 = 30; } @@ -498,11 +497,13 @@ struct EventCacheInner { /// See doc comment of [`EventCache::auto_shrink_linked_chunk_task`]. auto_shrink_sender: OnceLock>, - /// A sender for background requests, that is shared with every room. + /// A sender for automatic pagination requests, that is shared with every + /// room. /// /// It's a `OnceLock` because its initialization is deferred to /// [`EventCache::subscribe`]. - background_requests_sender: OnceLock>, + automatic_pagination_requests_sender: + OnceLock>, /// A sender for room generic update. /// @@ -705,7 +706,7 @@ impl EventCacheInner { "we must have called `EventCache::subscribe()` before calling here.", ), self.store.clone(), - self.background_requests_sender.get().cloned(), + self.automatic_pagination_requests_sender.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: From bdd01628313606a9a203f2f31d42ea1d4b6e02bc Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Mon, 30 Mar 2026 11:40:19 +0200 Subject: [PATCH 17/20] tests(event cache): make tests more resilient to the order of event cache updates --- .../src/event_cache/automatic_pagination.rs | 38 ++++++++++++++----- 1 file changed, 28 insertions(+), 10 deletions(-) diff --git a/crates/matrix-sdk/src/event_cache/automatic_pagination.rs b/crates/matrix-sdk/src/event_cache/automatic_pagination.rs index 931dd65b8..df40c5561 100644 --- a/crates/matrix-sdk/src/event_cache/automatic_pagination.rs +++ b/crates/matrix-sdk/src/event_cache/automatic_pagination.rs @@ -89,6 +89,7 @@ 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}; @@ -124,7 +125,17 @@ mod tests { let room_id = room_id!("!omelette:fromage.fr"); let f = EventFactory::new().room(room_id).sender(*BOB); - let room = server + 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) @@ -133,12 +144,14 @@ mod tests { ) .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()); + { + 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 @@ -210,9 +223,14 @@ mod tests { ) .await; - assert_let_timeout!( - Ok(RoomEventCacheUpdate::UpdateTimelineEvents(_)) = room_cache_updates.recv() - ); + { + 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 From 1ee88b176ce2cb69973bc83fe296d0cca5a06fc3 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Tue, 31 Mar 2026 15:41:13 +0200 Subject: [PATCH 18/20] refactor(event cache): create a standalone `AutomaticPagination` API object --- .../src/event_cache/automatic_pagination.rs | 101 ++++++++++++++---- .../matrix-sdk/src/event_cache/caches/mod.rs | 9 +- .../src/event_cache/caches/read_receipts.rs | 13 +-- .../src/event_cache/caches/room/state.rs | 21 ++-- crates/matrix-sdk/src/event_cache/mod.rs | 51 ++++----- 5 files changed, 113 insertions(+), 82 deletions(-) diff --git a/crates/matrix-sdk/src/event_cache/automatic_pagination.rs b/crates/matrix-sdk/src/event_cache/automatic_pagination.rs index df40c5561..38df2b06f 100644 --- a/crates/matrix-sdk/src/event_cache/automatic_pagination.rs +++ b/crates/matrix-sdk/src/event_cache/automatic_pagination.rs @@ -12,24 +12,86 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::{collections::HashMap, sync::Arc}; +use std::{ + collections::HashMap, + sync::{Arc, Weak}, +}; -use ruma::OwnedRoomId; +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_err() + } +} + +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)] -pub(crate) enum AutomaticPaginationRequest { +enum AutomaticPaginationRequest { PaginateRoomBackwards { room_id: OwnedRoomId }, } /// Listen to background automatic pagination requests, and execute them in /// real-time. #[instrument(skip_all)] -pub(super) async fn automatic_paginations_task( - inner: Arc, +async fn automatic_paginations_task( + inner: Weak, mut receiver: mpsc::UnboundedReceiver, ) { trace!("Spawning the automatic pagination task"); @@ -39,6 +101,11 @@ pub(super) async fn automatic_paginations_task( 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 @@ -93,25 +160,13 @@ mod tests { use matrix_sdk_base::sleep::sleep; use matrix_sdk_test::{BOB, JoinedRoomBuilder, async_test, event_factory::EventFactory}; use ruma::{event_id, room_id}; - use tokio::sync::mpsc; use crate::{ assert_let_timeout, - event_cache::{ - EventsOrigin, RoomEventCacheUpdate, - automatic_pagination::AutomaticPaginationRequest::PaginateRoomBackwards, - }, + event_cache::{EventsOrigin, RoomEventCacheUpdate}, test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate}, }; - impl super::super::EventCache { - fn pagination_requests_sender( - &self, - ) -> Option> { - self.inner.automatic_pagination_requests_sender.get().cloned() - } - } - /// Test that we can send automatic pagination requests. #[async_test] async fn test_background_room_paginations() { @@ -165,8 +220,8 @@ mod tests { .await; // Send a request for a background pagination, - let sender = event_cache.pagination_requests_sender().unwrap(); - sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); + 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!( @@ -247,8 +302,8 @@ mod tests { .await; // Send a request for a background pagination, - let sender = event_cache.pagination_requests_sender().unwrap(); - sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); + 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!( @@ -271,7 +326,7 @@ mod tests { assert!(room_cache_updates.is_empty()); // One can send another request to back-paginate… - sender.send(PaginateRoomBackwards { room_id: room_id.to_owned() }).unwrap(); + 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 diff --git a/crates/matrix-sdk/src/event_cache/caches/mod.rs b/crates/matrix-sdk/src/event_cache/caches/mod.rs index 56c5e9a97..3c15a7149 100644 --- a/crates/matrix-sdk/src/event_cache/caches/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/mod.rs @@ -25,8 +25,7 @@ use tokio::sync::{broadcast::Sender, mpsc}; use super::{EventCacheError, EventsOrigin, Result}; use crate::{ - client::WeakClient, event_cache::automatic_pagination::AutomaticPaginationRequest, - room::WeakRoom, + client::WeakClient, event_cache::automatic_pagination::AutomaticPagination, room::WeakRoom, }; pub mod event_focused; @@ -53,9 +52,7 @@ impl Caches { linked_chunk_update_sender: Sender, auto_shrink_sender: mpsc::Sender, store: EventCacheStoreLock, - automatic_pagination_requests_sender: Option< - mpsc::UnboundedSender, - >, + automatic_pagination: Option, ) -> Result { let Some(client) = weak_client.get() else { return Err(EventCacheError::ClientDropped); @@ -90,7 +87,7 @@ impl Caches { linked_chunk_update_sender, store, pagination_status.clone(), - automatic_pagination_requests_sender, + automatic_pagination, ) .await?; 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 f839a8118..9298070ef 100644 --- a/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs +++ b/crates/matrix-sdk/src/event_cache/caches/read_receipts.rs @@ -108,11 +108,10 @@ use ruma::{ }, serde::Raw, }; -use tokio::sync::mpsc::UnboundedSender; use tracing::{debug, instrument, trace, warn}; use crate::event_cache::{ - automatic_pagination::AutomaticPaginationRequest, caches::event_linked_chunk::EventLinkedChunk, + automatic_pagination::AutomaticPagination, caches::event_linked_chunk::EventLinkedChunk, }; trait RoomReadReceiptsExt { @@ -366,7 +365,7 @@ pub(crate) fn compute_unread_counts( linked_chunk: &EventLinkedChunk, read_receipts: &mut RoomReadReceipts, with_threading_support: bool, - automatic_pagination_request_sender: Option<&UnboundedSender>, + automatic_pagination: Option<&AutomaticPagination>, ) { debug!(?read_receipts, "Starting"); @@ -381,12 +380,8 @@ pub(crate) fn compute_unread_counts( if select_best_receipt_result.request_pagination { trace!("Requesting pagination to find a better receipt"); - if let Some(sender) = automatic_pagination_request_sender - && sender - .send(AutomaticPaginationRequest::PaginateRoomBackwards { - room_id: room_id.to_owned(), - }) - .is_err() + if let Some(automatic_pagination) = automatic_pagination + && !automatic_pagination.run_once(room_id) { warn!("Failed to request pagination to find a better receipt"); } 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 8678c92f5..260175d68 100644 --- a/crates/matrix-sdk/src/event_cache/caches/room/state.rs +++ b/crates/matrix-sdk/src/event_cache/caches/room/state.rs @@ -49,10 +49,7 @@ use ruma::{ room_version_rules::RoomVersionRules, serde::Raw, }; -use tokio::sync::{ - broadcast::{Receiver, Sender}, - mpsc, -}; +use tokio::sync::broadcast::{Receiver, Sender}; use tracing::{debug, error, instrument, trace, warn}; use super::{ @@ -77,8 +74,7 @@ use super::{ use crate::{ Room, event_cache::{ - automatic_pagination::AutomaticPaginationRequest, - caches::pagination::SharedPaginationStatus, + automatic_pagination::AutomaticPagination, caches::pagination::SharedPaginationStatus, }, room::WeakRoom, }; @@ -152,9 +148,8 @@ pub struct RoomEventCacheState { /// [`super::RoomEventCache`]. subscriber_count: Arc, - /// A sender to trigger automatic pagination requests under certain - /// predefined conditions. - automatic_pagination_request_sender: Option>, + /// A copy of the automatic pagination API object. + automatic_pagination: Option, } impl RoomEventCacheState { @@ -315,9 +310,7 @@ impl LockedRoomEventCacheState { linked_chunk_update_sender: Sender, store: EventCacheStoreLock, pagination_status: SharedObservable, - automatic_pagination_request_sender: Option< - mpsc::UnboundedSender, - >, + automatic_pagination: Option, ) -> Result { let store_guard = match store.lock().await? { // Lock is clean: all good! @@ -400,7 +393,7 @@ impl LockedRoomEventCacheState { waited_for_initial_prev_token: false, subscriber_count: Default::default(), pinned_event_cache: OnceLock::new(), - automatic_pagination_request_sender, + automatic_pagination, })) } } @@ -1064,7 +1057,7 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> { &self.state.room_linked_chunk, &mut read_receipts, self.state.enabled_thread_support, - self.state.automatic_pagination_request_sender.as_ref(), + 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 7d97d639f..297e115b1 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -53,7 +53,7 @@ use tracing::{error, instrument, trace}; use crate::{ Client, client::{ClientInner, WeakClient}, - event_cache::automatic_pagination::{AutomaticPaginationRequest, automatic_paginations_task}, + event_cache::automatic_pagination::AutomaticPagination, paginators::PaginatorError, }; @@ -154,9 +154,6 @@ pub struct EventCacheDropHandles { /// The task used to automatically shrink the linked chunks. auto_shrink_linked_chunk_task: BackgroundTaskHandle, - /// The task used to handle automatic pagination requests. - automatic_paginations_task: Option, - /// The task used to automatically redecrypt UTDs. #[cfg(feature = "e2e-encryption")] _redecryptor: redecryptor::Redecryptor, @@ -173,9 +170,6 @@ impl Drop for EventCacheDropHandles { self.listen_updates_task.abort(); self.ignore_user_list_update_task.abort(); self.auto_shrink_linked_chunk_task.abort(); - if let Some(task) = self.automatic_paginations_task.take() { - task.abort(); - } } } @@ -241,7 +235,6 @@ impl EventCache { by_room: Default::default(), drop_handles: Default::default(), auto_shrink_sender: Default::default(), - automatic_pagination_requests_sender: Default::default(), generic_update_sender, linked_chunk_update_sender, _thread_subscriber_task: thread_subscriber_task, @@ -251,6 +244,7 @@ impl EventCache { redecryption_channels, #[cfg(feature = "testing")] thread_subscriber_receiver: _thread_subscriber_receiver, + automatic_pagination: OnceLock::new(), }), } } @@ -320,21 +314,14 @@ impl EventCache { redecryptor::Redecryptor::new(&client, Arc::downgrade(&self.inner), receiver, &self.inner.linked_chunk_update_sender) }; - let automatic_paginations_task = if self.config().experimental_auto_backpagination { - let (sender, receiver) = mpsc::unbounded_channel(); - + if self.config().experimental_auto_backpagination { // Run the deferred initialization of the automatic pagination request sender, that // is shared with every room. - self.inner.automatic_pagination_requests_sender.get_or_init(|| sender); - - trace!("spawning the automatic paginations task"); - Some(task_monitor.spawn_background_task("event_cache::automatic_paginations_task", automatic_paginations_task( - self.inner.clone(), receiver - ))) + 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 task is disabled"); - None - }; + trace!("automatic paginations API is disabled"); + } Arc::new(EventCacheDropHandles { listen_updates_task, @@ -342,7 +329,6 @@ impl EventCache { auto_shrink_linked_chunk_task, #[cfg(feature = "e2e-encryption")] _redecryptor: redecryptor, - automatic_paginations_task }) }); @@ -393,6 +379,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. @@ -497,14 +490,6 @@ struct EventCacheInner { /// See doc comment of [`EventCache::auto_shrink_linked_chunk_task`]. auto_shrink_sender: OnceLock>, - /// A sender for automatic pagination requests, that is shared with every - /// room. - /// - /// It's a `OnceLock` because its initialization is deferred to - /// [`EventCache::subscribe`]. - automatic_pagination_requests_sender: - OnceLock>, - /// A sender for room generic update. /// /// See doc comment of [`RoomEventCacheGenericUpdate`] and @@ -547,6 +532,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; @@ -706,7 +697,7 @@ impl EventCacheInner { "we must have called `EventCache::subscribe()` before calling here.", ), self.store.clone(), - self.automatic_pagination_requests_sender.get().cloned(), + self.automatic_pagination.get().cloned(), ) .await?; From 1cf84e203e1fdee064fbd665c5e1618dd376fe9b Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Tue, 31 Mar 2026 15:57:19 +0200 Subject: [PATCH 19/20] chore(event cache): make clippy pass on automatic pagination --- crates/matrix-sdk/src/event_cache/automatic_pagination.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/crates/matrix-sdk/src/event_cache/automatic_pagination.rs b/crates/matrix-sdk/src/event_cache/automatic_pagination.rs index 38df2b06f..51c62cfa6 100644 --- a/crates/matrix-sdk/src/event_cache/automatic_pagination.rs +++ b/crates/matrix-sdk/src/event_cache/automatic_pagination.rs @@ -62,11 +62,10 @@ impl AutomaticPagination { // 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 + self.inner .sender .send(AutomaticPaginationRequest::PaginateRoomBackwards { room_id: room_id.to_owned() }) - .is_err() + .is_ok() } } From 06e6dd05c34539b7d56a4c0551287f9baf851c33 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Tue, 31 Mar 2026 16:36:13 +0200 Subject: [PATCH 20/20] chore(docs): publicly expose the `AutomaticPagination` API object for external users --- crates/matrix-sdk/src/event_cache/mod.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/matrix-sdk/src/event_cache/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index eb249d14e..2dedeb5a0 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -53,7 +53,6 @@ use tracing::{error, instrument, trace}; use crate::{ Client, client::{ClientInner, WeakClient}, - event_cache::automatic_pagination::AutomaticPagination, paginators::PaginatorError, }; @@ -79,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 {