From c3fc310f29d6d05271b07f5ced2695e7201f2528 Mon Sep 17 00:00:00 2001 From: Benjamin Bouvier Date: Wed, 19 Feb 2025 12:17:51 +0100 Subject: [PATCH] refactor(event cache): simplify back-pagination --- .../matrix-sdk-ui/src/timeline/pagination.rs | 54 ++--- .../integration/timeline/pinned_event.rs | 12 +- crates/matrix-sdk/CHANGELOG.md | 7 + crates/matrix-sdk/src/event_cache/mod.rs | 2 +- .../matrix-sdk/src/event_cache/pagination.rs | 115 ++++------ crates/matrix-sdk/src/event_cache/room/mod.rs | 16 +- .../tests/integration/event_cache.rs | 203 ++++++------------ 7 files changed, 130 insertions(+), 279 deletions(-) diff --git a/crates/matrix-sdk-ui/src/timeline/pagination.rs b/crates/matrix-sdk-ui/src/timeline/pagination.rs index 414572f44..61ca3593a 100644 --- a/crates/matrix-sdk-ui/src/timeline/pagination.rs +++ b/crates/matrix-sdk-ui/src/timeline/pagination.rs @@ -12,8 +12,6 @@ // 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; @@ -21,9 +19,9 @@ use futures_util::{pin_mut, StreamExt as _}; use matrix_sdk::event_cache::{ self, paginator::{PaginatorError, PaginatorState}, - BackPaginationOutcome, EventCacheError, RoomPagination, + EventCacheError, RoomPagination, }; -use tracing::{instrument, trace, warn}; +use tracing::{instrument, warn}; use super::Error; @@ -75,38 +73,30 @@ impl super::Timeline { /// /// Returns whether we hit the start of the timeline. async fn live_paginate_backwards(&self, batch_size: u16) -> event_cache::Result { - let pagination = self.event_cache.pagination(); - - let result = pagination - .run_backwards( - batch_size, - |BackPaginationOutcome { events, reached_start }, - _timeline_has_been_reset| async move { - let num_events = events.len(); - trace!("Back-pagination succeeded with {num_events} events"); - - 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(()); + loop { + match self.event_cache.pagination().run_backwards_once(batch_size).await { + Ok(outcome) => { + // As an exceptional contract, restart the back-pagination if we received an + // empty chunk. + if outcome.reached_start || !outcome.events.is_empty() { + return Ok(outcome.reached_start); } + } - ControlFlow::Break(reached_start) - }, - ) - .await; + Err(EventCacheError::BackpaginationError( + PaginatorError::InvalidPreviousState { + actual: PaginatorState::Paginating, .. + }, + )) => { + // Treat an already running pagination exceptionally, returning false so that + // the caller retries later. + warn!("Another pagination request is already happening, returning early"); + return Ok(false); + } - match result { - Err(EventCacheError::BackpaginationError(PaginatorError::InvalidPreviousState { - actual: PaginatorState::Paginating, - .. - })) => { - warn!("Another pagination request is already happening, returning early"); - Ok(false) + // Propagate other errors as such. + Err(err) => return Err(err), } - - result => result, } } 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 43e5b075c..140834ddd 100644 --- a/crates/matrix-sdk-ui/tests/integration/timeline/pinned_event.rs +++ b/crates/matrix-sdk-ui/tests/integration/timeline/pinned_event.rs @@ -1,11 +1,10 @@ -use std::{ops::ControlFlow, time::Duration}; +use std::time::Duration; use assert_matches2::assert_let; use eyeball_im::VectorDiff; use futures_util::StreamExt as _; use matrix_sdk::{ config::SyncSettings, - event_cache::{BackPaginationOutcome, TimelineHasBeenResetWhilePaginating}, test_utils::{ logged_in_client_with_server, mocks::{MatrixMockServer, RoomMessagesResponseTemplate}, @@ -428,17 +427,10 @@ async fn test_pinned_timeline_with_no_pinned_events_on_pagination_is_just_empty( let (event_cache, _) = room.event_cache().await.expect("Event cache should be accessible"); - async fn once( - outcome: BackPaginationOutcome, - _timeline_has_been_reset: TimelineHasBeenResetWhilePaginating, - ) -> ControlFlow { - ControlFlow::Break(outcome) - } - // Paginate backwards once using the event cache to load the event event_cache .pagination() - .run_backwards(10, once) + .run_backwards_once(10) .await .expect("Pagination of events should successful"); diff --git a/crates/matrix-sdk/CHANGELOG.md b/crates/matrix-sdk/CHANGELOG.md index 587994672..77f3f5abf 100644 --- a/crates/matrix-sdk/CHANGELOG.md +++ b/crates/matrix-sdk/CHANGELOG.md @@ -8,6 +8,13 @@ All notable changes to this project will be documented in this file. ### Features +- [**breaking**]: The `RoomPagination::run_backwards` method has been removed, and replaced by two +simpler methods: + - `RoomPagination::run_backwards_until()`, which will retrigger back-paginations until a certain + number of events have been received (and retry if the timeline has been reset in the background). + - `RoomPagination::run_backwards_once()`, which will run a single back-pagination (and retry if + the timeline has been reset in the background). + ([#4689](https://github.com/matrix-org/matrix-rust-sdk/pull/4689)) - [**breaking**]: The `Oidc::account_management_url` method now caches the result of a call, subsequent calls to the method will not contact the OIDC provider for a while, instead the cached URI will be returned. If caching of diff --git a/crates/matrix-sdk/src/event_cache/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index 12dfc3522..b3b2d94e8 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -69,7 +69,7 @@ mod pagination; mod room; pub mod paginator; -pub use pagination::{PaginationToken, RoomPagination, TimelineHasBeenResetWhilePaginating}; +pub use pagination::{PaginationToken, RoomPagination}; pub use room::RoomEventCache; /// An error observed in the [`EventCache`]. diff --git a/crates/matrix-sdk/src/event_cache/pagination.rs b/crates/matrix-sdk/src/event_cache/pagination.rs index e29c5d85a..9401b391b 100644 --- a/crates/matrix-sdk/src/event_cache/pagination.rs +++ b/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::{future::Future, ops::ControlFlow, sync::Arc, time::Duration}; +use std::{sync::Arc, time::Duration}; use eyeball::Subscriber; use matrix_sdk_base::timeout::timeout; @@ -45,76 +45,45 @@ 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 [`ControlFlow::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. - /// - /// # Example - /// - /// To do a single run: - /// - /// ```rust - /// use std::ops::ControlFlow; - /// - /// use matrix_sdk::event_cache::{ - /// BackPaginationOutcome, - /// RoomPagination, - /// TimelineHasBeenResetWhilePaginating - /// }; - /// - /// # async fn foo(room_pagination: RoomPagination) { - /// let result = room_pagination.run_backwards( - /// 42, - /// |BackPaginationOutcome { events, reached_start }, - /// _timeline_has_been_reset: TimelineHasBeenResetWhilePaginating| 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( + /// It will run multiple back-paginations until one of these two conditions + /// is met: + /// - either we've reached the start of the timeline, + /// - or we've obtained enough events to fulfill the requested number of + /// events. + #[instrument(skip(self))] + pub async fn run_backwards_until( &self, - batch_size: u16, - mut until: Until, - ) -> Result - where - Until: FnMut(BackPaginationOutcome, TimelineHasBeenResetWhilePaginating) -> UntilFuture, - UntilFuture: Future>, - { - let mut timeline_has_been_reset = TimelineHasBeenResetWhilePaginating::No; + num_requested_events: u16, + ) -> Result { + let mut events = Vec::new(); loop { - if let Some(outcome) = self.run_backwards_impl(batch_size).await? { - match until(outcome, timeline_has_been_reset).await { - ControlFlow::Continue(()) => { - trace!("back-pagination continues"); - - timeline_has_been_reset = TimelineHasBeenResetWhilePaginating::No; - - continue; - } - - ControlFlow::Break(value) => return Ok(value), + if let Some(outcome) = self.run_backwards_impl(num_requested_events).await? { + events.extend(outcome.events); + if outcome.reached_start || events.len() >= num_requested_events as usize { + return Ok(BackPaginationOutcome { + reached_start: outcome.reached_start, + events, + }); } + trace!("restarting back-pagination, because we haven't reached the start or obtained enough events yet"); } - timeline_has_been_reset = TimelineHasBeenResetWhilePaginating::Yes; + debug!("restarting back-pagination because of a timeline reset."); + } + } - debug!("back-pagination has been internally restarted because of a timeline reset."); + /// Run a single back-pagination for the requested number of events. + /// + /// This automatically takes care of waiting for a pagination token from + /// sync, if we haven't done that before. + #[instrument(skip(self))] + pub async fn run_backwards_once(&self, batch_size: u16) -> Result { + loop { + if let Some(outcome) = self.run_backwards_impl(batch_size).await? { + return Ok(outcome); + } + debug!("restarting back-pagination because of a timeline reset."); } } @@ -209,18 +178,16 @@ impl RoomPagination { // During a backwards pagination, when a duplicated event is found, the old // event is kept and the new event is ignored. This is the opposite strategy // than during a sync where the old event is removed and the new event is added. - // - // Let's forget the new events that are duplicated. if !all_deduplicated { + // Let's forget the new events that are duplicated. events.retain(|new_event| { new_event .event_id() .map(|event_id| !duplicated_event_ids.contains(&event_id)) .unwrap_or(false) }); - } - // All new events are duplicated, they can all be ignored. - else { + } else { + // All new events are duplicated, they can all be ignored. events.clear(); } @@ -418,16 +385,6 @@ impl From> for PaginationToken { } } -/// A type representing whether the timeline has been reset. -#[derive(Debug)] -pub enum TimelineHasBeenResetWhilePaginating { - /// The timeline has been reset. - Yes, - - /// The timeline has not been reset. - No, -} - #[cfg(test)] mod tests { // Those tests require time to work, and it does not on wasm32. diff --git a/crates/matrix-sdk/src/event_cache/room/mod.rs b/crates/matrix-sdk/src/event_cache/room/mod.rs index f9adadd38..b74533309 100644 --- a/crates/matrix-sdk/src/event_cache/room/mod.rs +++ b/crates/matrix-sdk/src/event_cache/room/mod.rs @@ -1290,8 +1290,6 @@ mod tests { #[cfg(not(target_arch = "wasm32"))] // This uses the cross-process lock, so needs time support. #[async_test] async fn test_clear() { - use std::ops::ControlFlow; - use eyeball_im::VectorDiff; use matrix_sdk_base::linked_chunk::LinkedChunkBuilderTest; @@ -1388,11 +1386,7 @@ mod tests { // Let's load more chunks to get all events. { - room_event_cache - .pagination() - .run_backwards(20, |outcome, _| async move { ControlFlow::Break(outcome) }) - .await - .unwrap(); + room_event_cache.pagination().run_backwards_once(20).await.unwrap(); assert_let_timeout!( Ok(RoomEventCacheUpdate::UpdateTimelineEvents { diffs, .. }) = stream.recv() @@ -1437,8 +1431,6 @@ mod tests { #[cfg(not(target_arch = "wasm32"))] // This uses the cross-process lock, so needs time support. #[async_test] async fn test_load_from_storage() { - use std::ops::ControlFlow; - use eyeball_im::VectorDiff; use super::RoomEventCacheUpdate; @@ -1530,11 +1522,7 @@ mod tests { assert!(room_event_cache.event(event_id2).await.is_some()); // Let's paginate to load more events. - room_event_cache - .pagination() - .run_backwards(20, |outcome, _| async move { ControlFlow::Break(outcome) }) - .await - .unwrap(); + room_event_cache.pagination().run_backwards_once(20).await.unwrap(); assert_let_timeout!( Ok(RoomEventCacheUpdate::UpdateTimelineEvents { diffs, .. }) = stream.recv() diff --git a/crates/matrix-sdk/tests/integration/event_cache.rs b/crates/matrix-sdk/tests/integration/event_cache.rs index 5437b4a03..f7d28e022 100644 --- a/crates/matrix-sdk/tests/integration/event_cache.rs +++ b/crates/matrix-sdk/tests/integration/event_cache.rs @@ -1,8 +1,4 @@ -use std::{ - future::ready, - ops::{ControlFlow, Not}, - time::Duration, -}; +use std::{ops::Not, time::Duration}; use assert_matches::assert_matches; use assert_matches2::assert_let; @@ -13,7 +9,7 @@ use matrix_sdk::{ deserialized_responses::TimelineEvent, event_cache::{ paginator::PaginatorState, BackPaginationOutcome, EventCacheError, PaginationToken, - RoomEventCacheUpdate, TimelineHasBeenResetWhilePaginating, + RoomEventCacheUpdate, }, linked_chunk::{ChunkIdentifier, Position, Update}, test_utils::{ @@ -33,13 +29,6 @@ use ruma::{ use serde_json::json; use tokio::{spawn, sync::broadcast, time::sleep}; -async fn once( - outcome: BackPaginationOutcome, - _timeline_has_been_reset: TimelineHasBeenResetWhilePaginating, -) -> ControlFlow { - ControlFlow::Break(outcome) -} - #[async_test] async fn test_must_explicitly_subscribe() { let server = MatrixMockServer::new().await; @@ -284,7 +273,7 @@ async fn test_backpaginate_once() { assert_matches!(pagination.get_or_wait_for_token(None).await, PaginationToken::HasMore(_)); - pagination.run_backwards(20, once).await.unwrap() + pagination.run_backwards_once(20).await.unwrap() }; // I'll get all the previous events, in "reverse" order (same as the response). @@ -351,7 +340,6 @@ async fn test_backpaginate_many_times_with_many_iterations() { wait_for_initial_events(events, &mut room_stream).await; let mut num_iterations = 0; - let mut num_paginations = 0; let mut global_events = Vec::new(); let mut global_reached_start = false; @@ -380,29 +368,18 @@ async fn test_backpaginate_many_times_with_many_iterations() { // Then if I backpaginate in a loop, let pagination = room_event_cache.pagination(); while matches!(pagination.get_or_wait_for_token(None).await, PaginationToken::HasMore(_)) { - pagination - .run_backwards(20, |outcome, timeline_has_been_reset| { - num_paginations += 1; + let outcome = pagination.run_backwards_once(20).await.unwrap(); - assert_matches!(timeline_has_been_reset, TimelineHasBeenResetWhilePaginating::No); - - if !global_reached_start { - global_reached_start = outcome.reached_start; - } - - global_events.extend(outcome.events); - - ready(ControlFlow::Break(())) - }) - .await - .unwrap(); + global_events.extend(outcome.events); + if !global_reached_start { + global_reached_start = outcome.reached_start; + } num_iterations += 1; } // I'll get all the previous events, - assert_eq!(num_iterations, 2); // in two iterations… - assert_eq!(num_paginations, 2); // … we get two paginations. + assert_eq!(num_iterations, 2); // in two iterations assert!(global_reached_start); assert_event_matches_msg(&global_events[0], "world"); @@ -489,7 +466,6 @@ async fn test_backpaginate_many_times_with_one_iteration() { wait_for_initial_events(events, &mut room_stream).await; let mut num_iterations = 0; - let mut num_paginations = 0; let mut global_events = Vec::new(); let mut global_reached_start = false; @@ -518,33 +494,16 @@ async fn test_backpaginate_many_times_with_one_iteration() { // Then if I backpaginate in a loop, let pagination = room_event_cache.pagination(); while matches!(pagination.get_or_wait_for_token(None).await, PaginationToken::HasMore(_)) { - pagination - .run_backwards(20, |outcome, timeline_has_been_reset| { - num_paginations += 1; - - assert_matches!(timeline_has_been_reset, TimelineHasBeenResetWhilePaginating::No); - - if !global_reached_start { - global_reached_start = outcome.reached_start; - } - - global_events.extend(outcome.events); - - ready(if outcome.reached_start { - ControlFlow::Break(()) - } else { - ControlFlow::Continue(()) - }) - }) - .await - .unwrap(); - + let outcome = pagination.run_backwards_until(20).await.unwrap(); + if !global_reached_start { + global_reached_start = outcome.reached_start; + } + global_events.extend(outcome.events); num_iterations += 1; } // I'll get all the previous events, - assert_eq!(num_iterations, 1); // in one iteration… - assert_eq!(num_paginations, 2); // … we get two paginations! + assert_eq!(num_iterations, 1); // in one iteration assert!(global_reached_start); assert_event_matches_msg(&global_events[0], "world"); @@ -669,18 +628,7 @@ async fn test_reset_while_backpaginating() { let backpagination = spawn({ let pagination = room_event_cache.pagination(); - async move { - pagination - .run_backwards(20, |outcome, timeline_has_been_reset| { - assert_matches!( - timeline_has_been_reset, - TimelineHasBeenResetWhilePaginating::Yes - ); - - ready(ControlFlow::Break(outcome)) - }) - .await - } + async move { pagination.run_backwards_once(20).await } }); // Receive the sync response (which clears the timeline). @@ -785,7 +733,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, once).await.unwrap(); + pagination.run_backwards_once(20).await.unwrap(); assert!(reached_start); @@ -844,7 +792,7 @@ async fn test_limited_timeline_resets_pagination() { // 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, once).await.unwrap(); + pagination.run_backwards_once(20).await.unwrap(); assert_eq!(events.len(), 1); assert!(reached_start); @@ -1011,7 +959,7 @@ async fn test_limited_timeline_without_storage() { .await; // We run back-pagination with success. - room_event_cache.pagination().run_backwards(20, once).await.unwrap(); + room_event_cache.pagination().run_backwards_once(20).await.unwrap(); // And we get the back-paginated event. assert_let_timeout!( @@ -1096,7 +1044,7 @@ async fn test_backpaginate_with_no_initial_events() { // timeline. let pagination_clone = pagination.clone(); - let first_pagination = spawn(async move { pagination_clone.run_backwards(20, once).await }); + let first_pagination = spawn(async move { pagination_clone.run_backwards_once(20).await }); // Make sure we've waited for the initial token long enough (3 seconds, as of // 2024-12-16). @@ -1112,7 +1060,7 @@ async fn test_backpaginate_with_no_initial_events() { first_pagination.await.expect("joining must work").expect("first backpagination must work"); // Second pagination will be instant. - pagination.run_backwards(20, once).await.unwrap(); + pagination.run_backwards_once(20).await.unwrap(); // The linked chunk should contain the events in the correct order. let (events, _stream) = room_event_cache.subscribe().await; @@ -1174,8 +1122,8 @@ async fn test_backpaginate_replace_empty_gap() { let pagination = room_event_cache.pagination(); // Run pagination twice. - pagination.run_backwards(20, once).await.unwrap(); - pagination.run_backwards(20, once).await.unwrap(); + pagination.run_backwards_once(20).await.unwrap(); + pagination.run_backwards_once(20).await.unwrap(); // The linked chunk should contain the events in the correct order. let (events, _stream) = room_event_cache.subscribe().await; @@ -1238,7 +1186,7 @@ async fn test_no_gap_stored_after_deduplicated_sync() { let pagination = room_event_cache.pagination(); // Run pagination once: it will consume the unique gap we had. - pagination.run_backwards(20, once).await.unwrap(); + pagination.run_backwards_once(20).await.unwrap(); // Now simulate that the sync returns the same events (which can happen with // simplified sliding sync). @@ -1256,7 +1204,7 @@ async fn test_no_gap_stored_after_deduplicated_sync() { // If this back-pagination fails, that's because we've stored a gap that's // useless. It should be short-circuited because there's no previous gap. - let outcome = pagination.run_backwards(20, once).await.unwrap(); + let outcome = pagination.run_backwards_once(20).await.unwrap(); assert!(outcome.reached_start); let (events, stream) = room_event_cache.subscribe().await; @@ -1366,7 +1314,7 @@ async fn test_no_gap_stored_after_deduplicated_backpagination() { // Run pagination once: it will consume prev-batch2 first, which is the most // recent token. - let outcome = pagination.run_backwards(20, once).await.unwrap(); + let outcome = pagination.run_backwards_once(20).await.unwrap(); // The pagination is empty: no new event. assert!(outcome.reached_start); @@ -1375,7 +1323,7 @@ async fn test_no_gap_stored_after_deduplicated_backpagination() { // Run pagination a second time: it will consume prev-batch, which is the least // recent token. - let outcome = pagination.run_backwards(20, once).await.unwrap(); + let outcome = pagination.run_backwards_once(20).await.unwrap(); // The pagination contains deduplicated events; they are all deduplicated; the // gap is replaced by zero event: nothing happens. @@ -1386,7 +1334,7 @@ async fn test_no_gap_stored_after_deduplicated_backpagination() { // If this back-pagination fails, that's because we've stored a gap that's // useless. It should be short-circuited because storing the previous gap was // useless. - let outcome = pagination.run_backwards(20, once).await.unwrap(); + let outcome = pagination.run_backwards_once(20).await.unwrap(); assert!(outcome.reached_start); assert!(outcome.events.is_empty()); assert!(stream.is_empty()); @@ -1443,7 +1391,7 @@ async fn test_dont_delete_gap_that_wasnt_inserted() { .mock_once() .mount() .await; - room_event_cache.pagination().run_backwards(20, once).await.unwrap(); + room_event_cache.pagination().run_backwards_once(20).await.unwrap(); // This doesn't cause an update, because nothing changed. assert!(stream.is_empty()); @@ -1623,17 +1571,6 @@ async fn test_lazy_loading() { let mock_server = MatrixMockServer::new().await; let client = mock_server.client_builder().build().await; - async fn until_at_least_one_event( - outcome: BackPaginationOutcome, - _timeline_has_been_reset: TimelineHasBeenResetWhilePaginating, - ) -> ControlFlow { - if outcome.reached_start || outcome.events.is_empty().not() { - ControlFlow::Break(outcome) - } else { - ControlFlow::Continue(()) - } - } - // Set up the event cache store. { let event_cache_store = client.event_cache_store().lock().await.unwrap(); @@ -1748,25 +1685,21 @@ async fn test_lazy_loading() { // One more chunk will be loaded from the store. This new chunk contains 5 // items. No need to reach the network. { - let pagination_outcome = room_event_cache - .pagination() - .run_backwards(10, until_at_least_one_event) - .await - .unwrap(); + let outcome = room_event_cache.pagination().run_backwards_until(1).await.unwrap(); // Oh! 5 events! How classy. - assert_eq!(pagination_outcome.events.len(), 5); + assert_eq!(outcome.events.len(), 5); // Hello you. Well… Uoy olleh! Remember, this is a backwards pagination, so // events are returned in reverse order. - assert_event_id!(pagination_outcome.events[0], "$ev2_4"); - assert_event_id!(pagination_outcome.events[1], "$ev2_3"); - assert_event_id!(pagination_outcome.events[2], "$ev2_2"); - assert_event_id!(pagination_outcome.events[3], "$ev2_1"); - assert_event_id!(pagination_outcome.events[4], "$ev2_0"); + assert_event_id!(outcome.events[0], "$ev2_4"); + assert_event_id!(outcome.events[1], "$ev2_3"); + assert_event_id!(outcome.events[2], "$ev2_2"); + assert_event_id!(outcome.events[3], "$ev2_1"); + assert_event_id!(outcome.events[4], "$ev2_0"); // And there is more, but this, kids, is for later. - assert!(pagination_outcome.reached_start.not()); + assert!(outcome.reached_start.not()); // Let's check the stream. It should reflect what the // `pagination_outcome` provides. @@ -1823,23 +1756,19 @@ async fn test_lazy_loading() { .mount_as_scoped() .await; - let pagination_outcome = room_event_cache - .pagination() - .run_backwards(10, until_at_least_one_event) - .await - .unwrap(); + let outcome = room_event_cache.pagination().run_backwards_until(1).await.unwrap(); // 🙈… 4 events! Of course. We've never doubt. - assert_eq!(pagination_outcome.events.len(), 4); + assert_eq!(outcome.events.len(), 4); // Hello you, in reverse order because this is a backward pagination. - assert_event_id!(pagination_outcome.events[0], "$ev1_4"); - assert_event_id!(pagination_outcome.events[1], "$ev1_3"); - assert_event_id!(pagination_outcome.events[2], "$ev1_2"); - assert_event_id!(pagination_outcome.events[3], "$ev1_1"); + assert_event_id!(outcome.events[0], "$ev1_4"); + assert_event_id!(outcome.events[1], "$ev1_3"); + assert_event_id!(outcome.events[2], "$ev1_2"); + assert_event_id!(outcome.events[3], "$ev1_1"); // And there is more because we didn't reach the start of the timeline yet. - assert!(pagination_outcome.reached_start.not()); + assert!(outcome.reached_start.not()); // Let's check the stream. It should reflect what the // `pagination_outcome` provides. @@ -1894,20 +1823,16 @@ async fn test_lazy_loading() { .mount_as_scoped() .await; - let pagination_outcome = room_event_cache - .pagination() - .run_backwards(10, until_at_least_one_event) - .await - .unwrap(); + let outcome = room_event_cache.pagination().run_backwards_until(1).await.unwrap(); // 🙊… 1 event! Indeed, `$ev0_5` has been filtered out. - assert_eq!(pagination_outcome.events.len(), 1); + assert_eq!(outcome.events.len(), 1); // Hello lonely. - assert_event_id!(pagination_outcome.events[0], "$ev1_0"); + assert_event_id!(outcome.events[0], "$ev1_0"); // Still not the start of the timeline. - assert!(pagination_outcome.reached_start.not()); + assert!(outcome.reached_start.not()); // Let's check the stream. // @@ -1961,29 +1886,25 @@ async fn test_lazy_loading() { .mount_as_scoped() .await; - let pagination_outcome = room_event_cache - .pagination() - .run_backwards(10, until_at_least_one_event) - .await - .unwrap(); + let outcome = room_event_cache.pagination().run_backwards_until(1).await.unwrap(); // 🙊 … 6 events! Wait, what? Yes! The network has returned 2 known events, they // have all been deduplicated, resulting in the removal of the gap chunk. // `until_at_least_one_event` re-runs the pagination, and this time the store is // reached. - assert_eq!(pagination_outcome.events.len(), 6); + assert_eq!(outcome.events.len(), 6); // Hello to all of you! - assert_event_id!(pagination_outcome.events[0], "$ev0_5"); - assert_event_id!(pagination_outcome.events[1], "$ev0_4"); - assert_event_id!(pagination_outcome.events[2], "$ev0_3"); - assert_event_id!(pagination_outcome.events[3], "$ev0_2"); - assert_event_id!(pagination_outcome.events[4], "$ev0_1"); - assert_event_id!(pagination_outcome.events[5], "$ev0_0"); + assert_event_id!(outcome.events[0], "$ev0_5"); + assert_event_id!(outcome.events[1], "$ev0_4"); + assert_event_id!(outcome.events[2], "$ev0_3"); + assert_event_id!(outcome.events[3], "$ev0_2"); + assert_event_id!(outcome.events[4], "$ev0_1"); + assert_event_id!(outcome.events[5], "$ev0_0"); // The start of the timeline isn't reached yet. What we know for the moment is // that we get new events. - assert!(pagination_outcome.reached_start.not()); + assert!(outcome.reached_start.not()); // Let's check the stream for the last time. let update = updates_stream.recv().await.unwrap(); @@ -2019,16 +1940,12 @@ async fn test_lazy_loading() { // This time, the first chunk is loaded and there is nothing else to do, no gap, // nothing. We've reached the start of the timeline! { - let pagination_outcome = room_event_cache - .pagination() - .run_backwards(10, until_at_least_one_event) - .await - .unwrap(); + let outcome = room_event_cache.pagination().run_backwards_until(1).await.unwrap(); // No events, hmmm… - assert!(pagination_outcome.events.is_empty()); + assert!(outcome.events.is_empty()); // … that's because the start of the timeline is finally reached! - assert!(pagination_outcome.reached_start); + assert!(outcome.reached_start); } }