Dima/matrix-rust-sdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(event cache): simplify back-pagination

Browse Source
This commit is contained in:
Benjamin Bouvier
2025-02-19 12:17:51 +01:00
parent b9c7ffe7c3
commit c3fc310f29
7 changed files with 130 additions and 279 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/matrix-sdk-ui/src/timeline/pagination.rs
+22 -32
View File
@@ -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<bool> {
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,
}
}
crates/matrix-sdk-ui/tests/integration/timeline/pinned_event.rs
+2 -10
View File
@@ -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<BackPaginationOutcome, ()> {
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");
crates/matrix-sdk/CHANGELOG.md
+7
View File
@@ -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
crates/matrix-sdk/src/event_cache/mod.rs
+1 -1
View File
@@ -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`].
crates/matrix-sdk/src/event_cache/pagination.rs
+36 -79
View File
@@ -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<Until, Break, UntilFuture>(
/// 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<Break>
where
Until: FnMut(BackPaginationOutcome, TimelineHasBeenResetWhilePaginating) -> UntilFuture,
UntilFuture: Future<Output = ControlFlow<Break, ()>>,
{
let mut timeline_has_been_reset = TimelineHasBeenResetWhilePaginating::No;
num_requested_events: u16,
) -> Result<BackPaginationOutcome> {
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<BackPaginationOutcome> {
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<Option<String>> 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.
crates/matrix-sdk/src/event_cache/room/mod.rs
+2 -14
View File
@@ -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()
crates/matrix-sdk/tests/integration/event_cache.rs
+60 -143
View File
@@ -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<BackPaginationOutcome, ()> {
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<BackPaginationOutcome, ()> {
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);
}
}