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?;