feat(sdk): Add RoomPagination::run_backwards(…, until).

This patch adds a new argument to `RoomPagination::run_backwards`:
`until`. It becomes:

    pub async fn run_backwards<F, B, Fut>(&self,
        batch_size: u16,
        mut until: F
    ) -> Result<B>
    where
        F: FnMut(BackPaginationOutcome) -> Fut,
        Fut: Future<Output = ControlFlow<B, ()>>,

The idea behind `until` is to run pagination _until_ `until` returns
`ControlFlow::Break`, otherwise it continues paginating.

This is useful is many scenearii (cf. the documentation). This is
also and primarily the first step to stop adding events directly from
the pagination, and starts adding events only and strictly only from
`event_cache::RoomEventCacheUpdate` (again, see the `TODO` in the
documentation). This is not done in this patch for the sake of ease
of review.

Author: Hywan

crates/matrix-sdk-ui/src/timeline/pagination.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::ops::ControlFlow;
+
 use async_rx::StreamExt as _;
 use async_stream::stream;
 use futures_core::Stream;
@@ -67,49 +69,45 @@ impl super::Timeline {
     pub async fn live_paginate_backwards(&self, batch_size: u16) -> event_cache::Result<bool> {
         let pagination = self.event_cache.pagination();
 
-        loop {
-            let result = pagination.run_backwards(batch_size).await;
-
-            let event_cache_outcome = match result {
-                Ok(outcome) => outcome,
-
-                Err(EventCacheError::BackpaginationError(
-                    PaginatorError::InvalidPreviousState {
-                        actual: PaginatorState::Paginating, ..
-                    },
-                )) => {
-                    warn!("Another pagination request is already happening, returning early");
-                    return Ok(false);
-                }
-
-                Err(err) => return Err(err),
-            };
-
-            let BackPaginationOutcome { events, reached_start } = event_cache_outcome;
-
-            let num_events = events.len();
-            trace!("Back-pagination succeeded with {num_events} events");
-
-            self.inner
-                .add_events_at(events, TimelineEnd::Front, RemoteEventOrigin::Pagination)
-                .await;
-
-            if reached_start {
-                return Ok(true);
+        let reached_start = pagination
+            .run_backwards(
+                batch_size,
+                |BackPaginationOutcome { events, reached_start }| async move {
+                    let num_events = events.len();
+                    trace!("Back-pagination succeeded with {num_events} events");
+
+                    // TODO(hywan): Remove, and let spread events via
+                    // `matrix_sdk::event_cache::RoomEventCacheUpdate` from
+                    // `matrix_sdk::event_cache::RoomPagination::run_backwards`.
+                    self.inner
+                        .add_events_at(events, TimelineEnd::Front, RemoteEventOrigin::Pagination)
+                        .await;
+
+                    if num_events == 0 && !reached_start {
+                        // As an exceptional contract: if there were no events in the response,
+                        // and we've not hit the start of the timeline, retry until we get
+                        // some events or reach the start of the timeline.
+                        return ControlFlow::Continue(());
+                    }
+
+                    ControlFlow::Break(reached_start)
+                },
+            )
+            .await;
+
+        match reached_start {
+            Ok(reached_start) => Ok(reached_start),
+
+            Err(EventCacheError::BackpaginationError(PaginatorError::InvalidPreviousState {
+                actual: PaginatorState::Paginating,
+                ..
+            })) => {
+                warn!("Another pagination request is already happening, returning early");
+                Ok(false)
             }
 
-            if num_events == 0 {
-                // As an exceptional contract: if there were no events in the response,
-                // and we've not hit the start of the timeline, retry until we get
-                // some events or reach the start of the timeline.
-                continue;
-            }
-
-            // Exit the inner loop, and ask for another limit.
-            break;
+            Err(err) => Err(err),
         }
-
-        Ok(false)
     }
 
     /// Subscribe to the back-pagination status of a live timeline.

crates/matrix-sdk/src/event_cache/pagination.rs

@@ -14,7 +14,7 @@
 
 //! A sub-object for running pagination tasks on a given room.
 
-use std::{sync::Arc, time::Duration};
+use std::{future::Future, ops::ControlFlow, sync::Arc, time::Duration};
 
 use eyeball::Subscriber;
 use matrix_sdk_base::deserialized_responses::SyncTimelineEvent;
@@ -59,17 +59,60 @@ impl RoomPagination {
     /// This automatically takes care of waiting for a pagination token from
     /// sync, if we haven't done that before.
     ///
+    /// The `until` argument is an async closure that returns a [`ControlFlow`]
+    /// to decide whether a new pagination must be run or not. It's helpful when
+    /// the server replies with e.g. a certain set of events, but we would like
+    /// more, or the event we are looking for isn't part of this set: in this
+    /// case, `until` returns [`Control::Continue`], otherwise it returns
+    /// [`ControlFlow::Break`]. `until` receives [`BackPaginationOutcome`] as
+    /// its sole argument.
+    ///
     /// # Errors
     ///
     /// It may return an error if the pagination token used during
     /// back-pagination has disappeared while we started the pagination. In
     /// that case, it's desirable to call the method again.
-    #[instrument(skip(self))]
-    pub async fn run_backwards(&self, batch_size: u16) -> Result<BackPaginationOutcome> {
+    ///
+    /// # Example
+    ///
+    /// To do a single run:
+    ///
+    /// ```rust
+    /// use std::ops::ControlFlow;
+    ///
+    /// use matrix_sdk::event_cache::{BackPaginationOutcome, RoomPagination};
+    ///
+    /// # async fn foo(room_pagination: RoomPagination) {
+    /// let result = room_pagination.run_backwards(
+    ///     42,
+    ///     |BackPaginationOutcome { events, reached_start }| async move {
+    ///         // Do something with `events` and `reached_start` maybe?
+    ///         let _ = events;
+    ///         let _ = reached_start;
+    ///
+    ///         ControlFlow::Break(())
+    ///     }
+    /// ).await;
+    /// # }
+    #[instrument(skip(self, until))]
+    pub async fn run_backwards<F, B, Fut>(&self, batch_size: u16, mut until: F) -> Result<B>
+    where
+        F: FnMut(BackPaginationOutcome) -> Fut,
+        Fut: Future<Output = ControlFlow<B, ()>>,
+    {
         loop {
-            if let Some(result) = self.run_backwards_impl(batch_size).await? {
-                return Ok(result);
+            if let Some(outcome) = self.run_backwards_impl(batch_size).await? {
+                match until(outcome).await {
+                    ControlFlow::Continue(()) => {
+                        debug!("back-pagination continues");
+
+                        continue;
+                    }
+
+                    ControlFlow::Break(value) => return Ok(value),
+                }
             }
+
             debug!("back-pagination has been internally restarted because of a timeline reset.");
         }
     }

crates/matrix-sdk/tests/integration/event_cache.rs

@@ -1,4 +1,4 @@
-use std::time::Duration;
+use std::{ops::ControlFlow, time::Duration};
 
 use assert_matches2::{assert_let, assert_matches};
 use matrix_sdk::{
@@ -24,6 +24,10 @@ use wiremock::{
 
 use crate::mock_sync;
 
+pub async fn once(outcome: BackPaginationOutcome) -> ControlFlow<BackPaginationOutcome, ()> {
+    ControlFlow::Break(outcome)
+}
+
 #[async_test]
 async fn test_must_explicitly_subscribe() {
     let (client, server) = logged_in_client_with_server().await;
@@ -362,7 +366,7 @@ async fn test_backpaginate_once() {
 
         assert!(pagination.get_or_wait_for_token().await.is_some());
 
-        pagination.run_backwards(20).await.unwrap()
+        pagination.run_backwards(20, once).await.unwrap()
     };
 
     // I'll get all the previous events, in "reverse" order (same as the response).
@@ -450,7 +454,7 @@ async fn test_backpaginate_multiple_iterations() {
     let pagination = room_event_cache.pagination();
     while pagination.get_or_wait_for_token().await.is_some() {
         let BackPaginationOutcome { reached_start, events } =
-            pagination.run_backwards(20).await.unwrap();
+            pagination.run_backwards(20, once).await.unwrap();
 
         if !global_reached_start {
             global_reached_start = reached_start;
@@ -586,7 +590,7 @@ async fn test_reset_while_backpaginating() {
 
     let backpagination = spawn({
         let pagination = room_event_cache.pagination();
-        async move { pagination.run_backwards(20).await }
+        async move { pagination.run_backwards(20, once).await }
     });
 
     // Receive the sync response (which clears the timeline).
@@ -656,7 +660,7 @@ async fn test_backpaginating_without_token() {
     // If we try to back-paginate with a token, it will hit the end of the timeline
     // and give us the resulting event.
     let BackPaginationOutcome { events, reached_start } =
-        pagination.run_backwards(20).await.unwrap();
+        pagination.run_backwards(20, once).await.unwrap();
 
     assert!(reached_start);