fix(event cache): don't ditch a previous-batch token when we didn't have initial events (#4936)

See #4891 that shows a case where we should have saved the
previous-batch token, and instead ditched it, in the previous version.

Changes include:
- moving the code deciding to keep or ditch the `previous-batch` token
into `append_events_locked`.
- tweak the condition to ditch, so that the `previous-batch` token is
ditched only if we didn't have events in the event cache in the first
place, in addition to having storage + the timeline not being marked as
limited explicitly.

Credits to @zecakeh for the test case.

Author: bnjbvr

crates/matrix-sdk/src/event_cache/room/mod.rs

@@ -380,21 +380,11 @@ impl RoomEventCacheInner {
             // Add all the events to the backend.
             trace!("adding new events");
 
-            // If we have storage, only keep the previous-batch token if we have a limited
-            // timeline. Otherwise, we know about all the events, and we don't need to
-            // back-paginate, so we wouldn't make use of the given previous-batch token.
-            //
-            // If we don't have storage, even if the timeline isn't limited, we may not have
-            // saved the previous events in any cache, so we should always be
-            // able to retrieve those.
-            let prev_batch =
-                if has_storage && !timeline.limited { None } else { timeline.prev_batch };
-
             let mut state = self.state.write().await;
             self.append_events_locked(
+                has_storage,
                 &mut state,
-                timeline.events,
-                prev_batch,
+                timeline,
                 ephemeral_events,
                 ambiguity_changes,
             )
@@ -438,10 +428,17 @@ impl RoomEventCacheInner {
         });
 
         // Push the new events.
+
+        // This method is only used when we don't have storage, and
+        // it's conservative to consider that this new timeline is "limited",
+        // since we don't know if we have a gap or not.
+        let has_storage = false;
+        let limited = true;
+
         self.append_events_locked(
+            has_storage,
             &mut state,
-            timeline_events,
-            prev_batch.clone(),
+            Timeline { limited, prev_batch, events: timeline_events },
             ephemeral_events,
             ambiguity_changes,
         )
@@ -456,28 +453,37 @@ impl RoomEventCacheInner {
     /// This is a private implementation. It must not be exposed publicly.
     async fn append_events_locked(
         &self,
+        has_storage: bool,
         state: &mut RoomEventCacheState,
-        timeline_events: Vec<TimelineEvent>,
-        prev_batch: Option<String>,
+        timeline: Timeline,
         ephemeral_events: Vec<Raw<AnySyncEphemeralRoomEvent>>,
         ambiguity_changes: BTreeMap<OwnedEventId, AmbiguityChange>,
     ) -> Result<()> {
-        if timeline_events.is_empty()
+        let mut prev_batch = timeline.prev_batch;
+        if timeline.events.is_empty()
             && prev_batch.is_none()
             && ephemeral_events.is_empty()
             && ambiguity_changes.is_empty()
         {
             return Ok(());
         }
 
+        // Ditch the previous-batch token if we have storage, the sync isn't limited and
+        // we've seen at least one event in the past. In this case (and only this one),
+        // we should definitely know what the head of the timeline is (either we
+        // know about all the events, or we have a gap somewhere).
+        if has_storage && !timeline.limited && state.events().events().next().is_some() {
+            prev_batch = None;
+        }
+
         let (
             DeduplicationOutcome {
                 all_events: events,
                 in_memory_duplicated_event_ids,
                 in_store_duplicated_event_ids,
             },
             all_duplicates,
-        ) = state.collect_valid_and_duplicated_events(timeline_events).await?;
+        ) = state.collect_valid_and_duplicated_events(timeline.events).await?;
 
         // During a sync, when a duplicated event is found, the old event is removed and
         // the new event is added.
@@ -532,11 +538,11 @@ impl RoomEventCacheInner {
 
             timeline_event_diffs.extend(new_timeline_event_diffs);
 
-            if prev_batch.is_some() && !all_duplicates {
-                // If there was a previous batch token, and there's at least one non-duplicated
-                // new event, unload the chunks so it only contains the last
-                // one; otherwise, there might be a valid gap in between, and
-                // observers may not render it (yet).
+            if timeline.limited && prev_batch.is_some() && !all_duplicates {
+                // If there was a previous batch token for a limited timeline, and there's at
+                // least one non-duplicated new event, unload the chunks so it
+                // only contains the last one; otherwise, there might be a valid
+                // gap in between, and observers may not render it (yet).
                 //
                 // We must do this *after* the above call to `.with_events_mut`, so the new
                 // events and gaps are properly persisted to storage.

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

@@ -2627,3 +2627,113 @@ async fn test_clear_all_rooms() {
         event_cache_store.load_last_chunk(sleeping_room_id).await.unwrap();
     assert!(maybe_last_chunk.is_none());
 }
+
+#[async_test]
+async fn test_sync_while_back_paginate() {
+    let server = MatrixMockServer::new().await;
+
+    let room_id = room_id!("!galette:saucisse.bzh");
+    let f = EventFactory::new().room(room_id).sender(user_id!("@ben:saucisse.bzh"));
+
+    // Previous batch of events which will be received via /messages, in reverse
+    // chronological order.
+    let prev_events = vec![
+        f.text_msg("messages3").event_id(event_id!("$messages3")).into_raw_timeline(),
+        f.text_msg("messages2").event_id(event_id!("$messages2")).into_raw_timeline(),
+        f.text_msg("messages1").event_id(event_id!("$messages1")).into_raw_timeline(),
+    ];
+
+    // Batch of events which will be received via /sync, in chronological
+    // order.
+    let sync_events = [
+        f.text_msg("sync1").event_id(event_id!("$sync1")).into_raw_timeline(),
+        f.text_msg("sync2").event_id(event_id!("$sync2")).into_raw_timeline(),
+        f.text_msg("sync3").event_id(event_id!("$sync3")).into_raw_timeline(),
+    ];
+
+    let state_memory_store = matrix_sdk_base::store::MemoryStore::new();
+    let store_config = StoreConfig::new("le_store".to_owned())
+        .event_cache_store(Arc::new(MemoryStore::new()))
+        .state_store(state_memory_store);
+
+    {
+        // First, initialize the sync so the client is aware of the room, in the state
+        // store.
+        let client = server.client_builder().store_config(store_config.clone()).build().await;
+        server.sync_joined_room(&client, room_id).await;
+    }
+
+    // Then, use a new client that will restore the state from the state store, and
+    // with an empty event cache store.
+    let client = server.client_builder().store_config(store_config).build().await;
+    let room = client.get_room(room_id).unwrap();
+
+    client.event_cache().subscribe().unwrap();
+    client.event_cache().enable_storage().unwrap();
+
+    let (room_event_cache, _drop_handles) = room.event_cache().await.unwrap();
+    let (initial_events, mut subscriber) = room_event_cache.subscribe().await;
+    assert!(initial_events.is_empty());
+
+    // Mock /messages in case we use the prev_batch token from sync.
+    server
+        .mock_room_messages()
+        .match_from("token-before-sync-from-sync")
+        .ok(RoomMessagesResponseTemplate::default()
+            .end_token("token-before-messages")
+            .events(prev_events))
+        .named("messages")
+        .mount()
+        .await;
+    // Mock /messages in case we use no token.
+    server
+        .mock_room_messages()
+        .ok(RoomMessagesResponseTemplate::default()
+            .end_token("token-before-sync-from-messages")
+            .events(sync_events.clone().into_iter().rev().collect()))
+        .named("messages")
+        .mount()
+        .await;
+
+    // Spawn back pagination.
+    let pagination = room_event_cache.pagination();
+    let back_pagination_handle =
+        spawn(async move { pagination.run_backwards_once(3).await.unwrap() });
+
+    // Receive a non-limited sync while back pagination is happening.
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .set_timeline_prev_batch("token-before-sync-from-sync")
+                .add_timeline_bulk(sync_events.into_iter().map(ruma::serde::Raw::cast)),
+        )
+        .await;
+
+    assert_let_timeout!(
+        Ok(RoomEventCacheUpdate::UpdateTimelineEvents { diffs, .. }) = subscriber.recv()
+    );
+    assert_eq!(diffs.len(), 1);
+    assert_let!(VectorDiff::Append { values } = &diffs[0]);
+
+    assert_eq!(values.len(), 3);
+    assert_event_matches_msg(&values[0], "sync1");
+    assert_event_matches_msg(&values[1], "sync2");
+    assert_event_matches_msg(&values[2], "sync3");
+
+    // Back pagination should succeed, and we don't have reached the start.
+    let outcome = back_pagination_handle.await.unwrap();
+    assert!(outcome.reached_start.not());
+    assert_eq!(outcome.events.len(), 3);
+
+    // And the back-paginated events come down from the subscriber too.
+    assert_let_timeout!(
+        Ok(RoomEventCacheUpdate::UpdateTimelineEvents { diffs, .. }) = subscriber.recv()
+    );
+    assert_eq!(diffs.len(), 3);
+    assert_let!(VectorDiff::Insert { index: 0, value: _ } = &diffs[0]);
+    assert_let!(VectorDiff::Insert { index: 1, value: _ } = &diffs[1]);
+    assert_let!(VectorDiff::Insert { index: 2, value: _ } = &diffs[2]);
+
+    assert!(subscriber.is_empty());
+}