refactor(event cache): create a standalone AutomaticPagination API object

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<AutomaticPaginationInner>,
+}
+
+#[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<EventCacheInner>, 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<AutomaticPaginationRequest>,
+}
+
 #[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<EventCacheInner>,
+async fn automatic_paginations_task(
+    inner: Weak<EventCacheInner>,
     mut receiver: mpsc::UnboundedReceiver<AutomaticPaginationRequest>,
 ) {
     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<mpsc::UnboundedSender<super::AutomaticPaginationRequest>> {
-            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

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<room::RoomEventCacheLinkedChunkUpdate>,
         auto_shrink_sender: mpsc::Sender<OwnedRoomId>,
         store: EventCacheStoreLock,
-        automatic_pagination_requests_sender: Option<
-            mpsc::UnboundedSender<AutomaticPaginationRequest>,
-        >,
+        automatic_pagination: Option<AutomaticPagination>,
     ) -> Result<Self> {
         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?;
 

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<AutomaticPaginationRequest>>,
+    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");
         }

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<AtomicUsize>,
 
-    /// A sender to trigger automatic pagination requests under certain
-    /// predefined conditions.
-    automatic_pagination_request_sender: Option<mpsc::UnboundedSender<AutomaticPaginationRequest>>,
+    /// A copy of the automatic pagination API object.
+    automatic_pagination: Option<AutomaticPagination>,
 }
 
 impl RoomEventCacheState {
@@ -315,9 +310,7 @@ impl LockedRoomEventCacheState {
         linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
         store: EventCacheStoreLock,
         pagination_status: SharedObservable<SharedPaginationStatus>,
-        automatic_pagination_request_sender: Option<
-            mpsc::UnboundedSender<AutomaticPaginationRequest>,
-        >,
+        automatic_pagination: Option<AutomaticPagination>,
     ) -> Result<Self, EventCacheError> {
         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 {

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<BackgroundTaskHandle>,
-
     /// 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<RoomEventCacheGenericUpdate> {
         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<AutomaticPagination> {
+        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<mpsc::Sender<AutoShrinkChannelPayload>>,
 
-    /// 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<mpsc::UnboundedSender<AutomaticPaginationRequest>>,
-
     /// 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<AutomaticPagination>,
 }
 
 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?;