timeline: after an event cache update lag, fetch previous events back…

… from the cache

Fixes #3311.

When the timeline is lagging behind the event cache, it should not only
clear what it contains (because it may be lagging some information
coming from the event cache), it should also retrieve the events the
cache knows about, and adds them as if they were "initial" events.

This makes sure that the event cache's events and the timeline's events
are always the same, and that, in the case of a lag, there won't be any
missing chunks (caused by the event cache back-pagination being further
away in the past, than what's displayed in the timeline).

The lag behind a bit too hard to reproduce, I've not included a test
here; but I think this should be the right move.

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

@@ -21,10 +21,7 @@ use matrix_sdk::{
     executor::spawn,
     Room,
 };
-use ruma::{
-    events::{receipt::ReceiptType, AnySyncTimelineEvent},
-    RoomVersionId,
-};
+use ruma::{events::AnySyncTimelineEvent, RoomVersionId};
 use tokio::sync::{broadcast, mpsc};
 use tracing::{info, info_span, trace, warn, Instrument, Span};
 
@@ -35,7 +32,7 @@ use super::{
     queue::send_queued_messages,
     BackPaginationStatus, Timeline, TimelineDropHandle,
 };
-use crate::{timeline::inner::TimelineEnd, unable_to_decrypt_hook::UtdHookManager};
+use crate::unable_to_decrypt_hook::UtdHookManager;
 
 /// Builder that allows creating and configuring various parts of a
 /// [`Timeline`].
@@ -137,26 +134,16 @@ impl TimelineBuilder {
         let (events, mut event_subscriber) = room_event_cache.subscribe().await?;
 
         let has_events = !events.is_empty();
-        let track_read_marker_and_receipts = settings.track_read_receipts;
-
-        let mut inner = TimelineInner::new(room, unable_to_decrypt_hook).with_settings(settings);
 
-        if track_read_marker_and_receipts {
-            inner.populate_initial_user_receipt(ReceiptType::Read).await;
-            inner.populate_initial_user_receipt(ReceiptType::ReadPrivate).await;
-        }
+        let inner = TimelineInner::new(room, unable_to_decrypt_hook).with_settings(settings);
 
-        if has_events {
-            inner.add_events_at(events, TimelineEnd::Back { from_cache: true }).await;
-        }
-        if track_read_marker_and_receipts {
-            inner.load_fully_read_event().await;
-        }
+        inner.replace_with_initial_events(events).await;
 
         let room = inner.room();
         let client = room.client();
 
         let room_update_join_handle = spawn({
+            let room_event_cache = room_event_cache.clone();
             let inner = inner.clone();
 
             let span =
@@ -177,7 +164,23 @@ impl TimelineBuilder {
                                 num_skipped,
                                 "Lagged behind event cache updates, resetting timeline"
                             );
-                            inner.clear().await;
+
+                            // The updates might have lagged, but the room event cache might have
+                            // events, so retrieve them and add them back again to the timeline,
+                            // after clearing it.
+                            //
+                            // If we can't get a handle on the room cache's events, just clear the
+                            // current timeline.
+                            match room_event_cache.subscribe().await {
+                                Ok((events, _)) => {
+                                    inner.replace_with_initial_events(events).await;
+                                }
+                                Err(err) => {
+                                    warn!("Error when re-inserting initial events into the timeline: {err}");
+                                    inner.clear().await;
+                                }
+                            }
+
                             continue;
                         }
                     };

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

@@ -384,7 +384,7 @@ impl<P: RoomDataProvider> TimelineInner<P> {
 
     /// Populates our own latest read receipt in the in-memory by-user read
     /// receipt cache.
-    pub(super) async fn populate_initial_user_receipt(&mut self, receipt_type: ReceiptType) {
+    pub(super) async fn populate_initial_user_receipt(&self, receipt_type: ReceiptType) {
         let own_user_id = self.room_data_provider.own_user_id().to_owned();
 
         let mut read_receipt = self
@@ -965,6 +965,38 @@ impl TimelineInner {
         &self.room_data_provider
     }
 
+    /// Replaces the content of the current timeline with initial events.
+    ///
+    /// Also sets up read receipts and the read marker for a live timeline of a
+    /// room.
+    ///
+    /// This is all done with a single lock guard, since we don't want the state
+    /// to be modified between the clear and re-insertion of new events.
+    pub(super) async fn replace_with_initial_events(&self, events: Vec<SyncTimelineEvent>) {
+        let mut state = self.state.write().await;
+
+        state.clear();
+
+        let track_read_markers = self.settings.track_read_receipts;
+        if track_read_markers {
+            self.populate_initial_user_receipt(ReceiptType::Read).await;
+            self.populate_initial_user_receipt(ReceiptType::ReadPrivate).await;
+        }
+        if !events.is_empty() {
+            state
+                .add_events_at(
+                    events,
+                    TimelineEnd::Back { from_cache: true },
+                    &self.room_data_provider,
+                    &self.settings,
+                )
+                .await;
+        }
+        if track_read_markers {
+            self.load_fully_read_event().await;
+        }
+    }
+
     /// Get the current fully-read event, from storage.
     pub(super) async fn fully_read_event(&self) -> Option<FullyReadEvent> {
         match self.room().account_data_static().await {