event cache: reuse the paginator internally

Fixes #3355.

Author: bnjbvr

bindings/matrix-sdk-ffi/src/timeline/mod.rs

@@ -18,11 +18,14 @@ use anyhow::{Context, Result};
 use as_variant::as_variant;
 use eyeball_im::VectorDiff;
 use futures_util::{pin_mut, StreamExt};
-use matrix_sdk::attachment::{
-    AttachmentConfig, AttachmentInfo, BaseAudioInfo, BaseFileInfo, BaseImageInfo,
-    BaseThumbnailInfo, BaseVideoInfo, Thumbnail,
+use matrix_sdk::{
+    attachment::{
+        AttachmentConfig, AttachmentInfo, BaseAudioInfo, BaseFileInfo, BaseImageInfo,
+        BaseThumbnailInfo, BaseVideoInfo, Thumbnail,
+    },
+    event_cache::paginator::PaginatorState,
 };
-use matrix_sdk_ui::timeline::{EventItemOrigin, PaginationStatus, Profile, TimelineDetails};
+use matrix_sdk_ui::timeline::{EventItemOrigin, Profile, TimelineDetails};
 use mime::Mime;
 use ruma::{
     events::{
@@ -160,7 +163,7 @@ impl Timeline {
         self.inner.fetch_members().await
     }
 
-    pub fn subscribe_to_back_pagination_status(
+    pub async fn subscribe_to_back_pagination_status(
         &self,
         listener: Box<dyn PaginationStatusListener>,
     ) -> Result<Arc<TaskHandle>, ClientError> {
@@ -588,7 +591,7 @@ pub trait TimelineListener: Sync + Send {
 
 #[uniffi::export(callback_interface)]
 pub trait PaginationStatusListener: Sync + Send {
-    fn on_update(&self, status: PaginationStatus);
+    fn on_update(&self, status: PaginatorState);
 }
 
 #[derive(Clone, uniffi::Object)]

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

@@ -14,7 +14,6 @@
 
 use std::{collections::BTreeSet, sync::Arc};
 
-use eyeball::SharedObservable;
 use futures_util::{pin_mut, StreamExt};
 use matrix_sdk::{event_cache::RoomEventCacheUpdate, executor::spawn, Room};
 use ruma::{events::AnySyncTimelineEvent, RoomVersionId};
@@ -28,10 +27,7 @@ use super::{
     queue::send_queued_messages,
     Error, Timeline, TimelineDropHandle, TimelineFocus,
 };
-use crate::{
-    timeline::{event_item::RemoteEventOrigin, PaginationStatus},
-    unable_to_decrypt_hook::UtdHookManager,
-};
+use crate::{timeline::event_item::RemoteEventOrigin, unable_to_decrypt_hook::UtdHookManager};
 
 /// Builder that allows creating and configuring various parts of a
 /// [`Timeline`].
@@ -314,7 +310,6 @@ impl TimelineBuilder {
 
         let timeline = Timeline {
             inner,
-            back_pagination_status: SharedObservable::new(PaginationStatus::Idle),
             msg_sender,
             event_cache: room_event_cache,
             drop_handle: Arc::new(TimelineDropHandle {

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

@@ -18,7 +18,6 @@
 
 use std::{pin::Pin, sync::Arc, task::Poll};
 
-use eyeball::SharedObservable;
 use eyeball_im::VectorDiff;
 use futures_core::Stream;
 use imbl::Vector;
@@ -96,7 +95,7 @@ pub use self::{
     event_type_filter::TimelineEventTypeFilter,
     inner::default_event_filter,
     item::{TimelineItem, TimelineItemKind},
-    pagination::{PaginationOptions, PaginationOutcome, PaginationStatus},
+    pagination::{PaginationOptions, PaginationOutcome},
     polls::PollResult,
     reactions::ReactionSenderData,
     sliding_sync_ext::SlidingSyncRoomExt,
@@ -130,9 +129,6 @@ pub struct Timeline {
 
     /// References to long-running tasks held by the timeline.
     drop_handle: Arc<TimelineDropHandle>,
-
-    /// Observable for whether a backward pagination is currently running.
-    pub(crate) back_pagination_status: SharedObservable<PaginationStatus>,
 }
 
 // Implements hash etc

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

@@ -12,38 +12,19 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{fmt, ops::ControlFlow, sync::Arc, time::Duration};
-
-use eyeball::{SharedObservable, Subscriber};
-use matrix_sdk::event_cache::{self, BackPaginationOutcome};
+use std::{fmt, ops::ControlFlow, sync::Arc};
+
+use eyeball::Subscriber;
+use matrix_sdk::event_cache::{
+    self,
+    paginator::{PaginatorError, PaginatorState},
+    BackPaginationOutcome, EventCacheError,
+};
 use tracing::{instrument, trace, warn};
 
 use super::Error;
 use crate::timeline::{event_item::RemoteEventOrigin, inner::TimelineEnd};
 
-struct ResetStatusGuard {
-    status: SharedObservable<PaginationStatus>,
-    target: Option<PaginationStatus>,
-}
-
-impl ResetStatusGuard {
-    fn new(status: SharedObservable<PaginationStatus>, target: PaginationStatus) -> Self {
-        Self { status, target: Some(target) }
-    }
-
-    fn disarm(mut self) {
-        self.target = None;
-    }
-}
-
-impl Drop for ResetStatusGuard {
-    fn drop(&mut self) {
-        if let Some(target) = self.target.take() {
-            self.status.set_if_not_eq(target);
-        }
-    }
-}
-
 impl super::Timeline {
     /// Add more events to the start of the timeline.
     ///
@@ -88,34 +69,27 @@ impl super::Timeline {
         &self,
         mut options: PaginationOptions<'_>,
     ) -> event_cache::Result<bool> {
-        let back_pagination_status = &self.back_pagination_status;
-
-        if back_pagination_status.get() == PaginationStatus::TimelineEndReached {
-            warn!("Start of timeline reached, ignoring backwards-pagination request");
-            return Ok(true);
-        }
-
-        if back_pagination_status.set_if_not_eq(PaginationStatus::Paginating).is_none() {
-            warn!("Another back-pagination is already running in the background");
-            return Ok(false);
-        }
-
-        let reset_status_guard =
-            ResetStatusGuard::new(back_pagination_status.clone(), PaginationStatus::Idle);
-
-        // The first time, we allow to wait a bit for *a* back-pagination token to come
-        // over via sync.
-        const WAIT_FOR_TOKEN_TIMEOUT: Duration = Duration::from_secs(3);
-
-        let mut token =
-            self.event_cache.oldest_backpagination_token(Some(WAIT_FOR_TOKEN_TIMEOUT)).await?;
-
         let initial_options = options.clone();
         let mut outcome = PaginationOutcome::default();
 
         while let Some(batch_size) = options.next_event_limit(outcome) {
             loop {
-                match self.event_cache.backpaginate(batch_size, token).await? {
+                let result = self.event_cache.paginate_backwards(batch_size).await;
+
+                let event_cache_outcome = match result {
+                    Ok(outcome) => outcome,
+
+                    Err(EventCacheError::BackpaginationError(
+                        PaginatorError::InvalidPreviousState { actual, .. },
+                    )) if actual == PaginatorState::Paginating => {
+                        warn!("Another pagination request is already happening, returning early");
+                        return Ok(false);
+                    }
+
+                    Err(err) => return Err(err),
+                };
+
+                match event_cache_outcome {
                     BackPaginationOutcome::Success { events, reached_start } => {
                         let num_events = events.len();
                         trace!("Back-pagination succeeded with {num_events} events");
@@ -130,11 +104,6 @@ impl super::Timeline {
                             .await;
 
                         if reached_start {
-                            // Don't reset the status to `Idle`…
-                            reset_status_guard.disarm();
-                            // …and set it to `TimelineEndReached` instead.
-                            back_pagination_status
-                                .set_if_not_eq(PaginationStatus::TimelineEndReached);
                             return Ok(true);
                         }
 
@@ -148,8 +117,8 @@ impl super::Timeline {
 
                         if num_events == 0 {
                             // As an exceptional contract: if there were no events in the response,
-                            // see if we had another back-pagination token, and retry the request.
-                            token = self.event_cache.oldest_backpagination_token(None).await?;
+                            // and we've not hit the start of the timeline, so retry until we get
+                            // some events.
                             continue;
                         }
                     }
@@ -163,22 +132,20 @@ impl super::Timeline {
                     }
                 }
 
-                // Retrieve the next earliest back-pagination token.
-                token = self.event_cache.oldest_backpagination_token(None).await?;
-
                 // Exit the inner loop, and ask for another limit.
                 break;
             }
         }
 
-        // The status is automatically reset to idle by `reset_status_guard`.
-
         Ok(false)
     }
 
     /// Subscribe to the back-pagination status of the timeline.
-    pub fn back_pagination_status(&self) -> Subscriber<PaginationStatus> {
-        self.back_pagination_status.subscribe()
+    ///
+    /// Note: this may send multiple Paginating/Idle sequences during a single
+    /// call to [`Self::paginate_backwards()`].
+    pub fn back_pagination_status(&self) -> Subscriber<PaginatorState> {
+        self.event_cache.pagination_status()
     }
 }
 
@@ -330,19 +297,6 @@ pub struct PaginationOutcome {
     pub total_items_updated: u64,
 }
 
-/// The status of a pagination.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))]
-pub enum PaginationStatus {
-    /// No pagination happening.
-    Idle,
-    /// Timeline is paginating for this end.
-    Paginating,
-    /// An end of the timeline (front or back) has been reached by this
-    /// pagination.
-    TimelineEndReached,
-}
-
 #[cfg(test)]
 mod tests {
     use std::{

crates/matrix-sdk-ui/tests/integration/timeline/pagination.rs

@@ -16,17 +16,16 @@ use std::{sync::Arc, time::Duration};
 
 use assert_matches2::assert_let;
 use eyeball_im::VectorDiff;
-use futures_util::{
-    future::{join, join3},
-    FutureExt,
+use futures_util::future::{join, join3};
+use matrix_sdk::{
+    config::SyncSettings, event_cache::paginator::PaginatorState,
+    test_utils::logged_in_client_with_server,
 };
-use matrix_sdk::{config::SyncSettings, test_utils::logged_in_client_with_server};
 use matrix_sdk_test::{
     async_test, EventBuilder, JoinedRoomBuilder, StateTestEvent, SyncResponseBuilder, ALICE, BOB,
 };
 use matrix_sdk_ui::timeline::{
-    AnyOtherFullStateEventContent, PaginationOptions, PaginationStatus, RoomExt,
-    TimelineItemContent,
+    AnyOtherFullStateEventContent, PaginationOptions, RoomExt, TimelineItemContent,
 };
 use once_cell::sync::Lazy;
 use ruma::{
@@ -81,7 +80,7 @@ async fn test_back_pagination() {
         server.reset().await;
     };
     let observe_paginating = async {
-        assert_eq!(back_pagination_status.next().await, Some(PaginationStatus::Paginating));
+        assert_eq!(back_pagination_status.next().await, Some(PaginatorState::Paginating));
     };
     join(paginate, observe_paginating).await;
 
@@ -136,8 +135,10 @@ async fn test_back_pagination() {
         .mount(&server)
         .await;
 
-    timeline.live_paginate_backwards(PaginationOptions::simple_request(10)).await.unwrap();
-    assert_next_eq!(back_pagination_status, PaginationStatus::TimelineEndReached);
+    let hit_start =
+        timeline.live_paginate_backwards(PaginationOptions::simple_request(10)).await.unwrap();
+    assert!(hit_start);
+    assert_next_eq!(back_pagination_status, PaginatorState::Idle);
 }
 
 #[async_test]
@@ -274,8 +275,8 @@ async fn test_wait_for_token() {
             .unwrap();
     };
     let observe_paginating = async {
-        assert_eq!(back_pagination_status.next().await, Some(PaginationStatus::Paginating));
-        assert_eq!(back_pagination_status.next().await, Some(PaginationStatus::Idle));
+        assert_eq!(back_pagination_status.next().await, Some(PaginatorState::Paginating));
+        assert_eq!(back_pagination_status.next().await, Some(PaginatorState::Idle));
     };
     let sync = async {
         // Make sure syncing starts a little bit later than pagination
@@ -430,18 +431,41 @@ async fn test_timeline_reset_while_paginating() {
         timeline
             .live_paginate_backwards(PaginationOptions::simple_request(10).wait_for_token())
             .await
-            .unwrap();
+            .unwrap()
     };
+
     let observe_paginating = async {
-        assert_eq!(back_pagination_status.next().await, Some(PaginationStatus::Paginating));
-        // timeline start reached because second pagination response contains
-        // no end field
-        assert_eq!(back_pagination_status.next().await, Some(PaginationStatus::TimelineEndReached));
+        let mut seen_paginating = false;
+
+        // Observe paginating updates: we want to make sure we see at least once
+        // Paginating, and that it settles with Idle.
+        while let Ok(update) =
+            timeout(Duration::from_millis(500), back_pagination_status.next()).await
+        {
+            match update {
+                Some(state) => {
+                    if state == PaginatorState::Paginating {
+                        seen_paginating = true;
+                    }
+                }
+                None => break,
+            }
+        }
+
+        assert!(seen_paginating);
+        assert_eq!(back_pagination_status.next_now(), PaginatorState::Idle);
     };
+
     let sync = async {
         client.sync_once(sync_settings.clone()).await.unwrap();
     };
-    timeout(Duration::from_secs(2), join3(paginate, observe_paginating, sync)).await.unwrap();
+
+    let (hit_start, _, _) =
+        timeout(Duration::from_secs(5), join3(paginate, observe_paginating, sync)).await.unwrap();
+
+    // timeline start reached because second pagination response contains no end
+    // field.
+    assert!(hit_start);
 
     // No events in back-pagination responses, day divider + event from latest
     // sync is present
@@ -570,7 +594,7 @@ async fn test_empty_chunk() {
         server.reset().await;
     };
     let observe_paginating = async {
-        assert_eq!(back_pagination_status.next().await, Some(PaginationStatus::Paginating));
+        assert_eq!(back_pagination_status.next().await, Some(PaginatorState::Paginating));
     };
     join(paginate, observe_paginating).await;
 
@@ -669,7 +693,7 @@ async fn test_until_num_items_with_empty_chunk() {
         server.reset().await;
     };
     let observe_paginating = async {
-        assert_eq!(back_pagination_status.next().await, Some(PaginationStatus::Paginating));
+        assert_eq!(back_pagination_status.next().await, Some(PaginatorState::Paginating));
     };
     join(paginate, observe_paginating).await;