matrix-rust-sdk

public

Star0

Fork 0

Watch0

20 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `3c85630430e1`

tarball

atagen <boss@atagen.co>2026-10-25 08:58 UTC

6 files changed+141-52

event-cache: keep pinned and event-focused views alive across a clear

clear_loaded_views dropped the PinnedEventCache and drained the EventFocusedCache map, so an open pinned timeline got a Clear and then a closed stream and an event-focused one got MissingCache on its next paginate. Both now empty in place like threads and refill themselves, the pinned cache by restarting its listener and the focused cache by running its /context start again behind the generation fence. A pagination fenced by a clear no longer claims hit_end_of_timeline, and the timeline Lagged paths swap in the receiver they resubscribed with.

Modified file: crates/matrix-sdk-ui/src/timeline/tasks.rs+14 additions−12 deletions@@ -56,18 +56,19 @@ pub(in crate::timeline) async fn pinned_events_task(
                 // 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.
-                let (initial_events, _) = match room_event_cache.subscribe_to_pinned_events().await
-                {
-                    Ok(initial_events) => initial_events,
-                    Err(err) => {
-                        error!(
-                            ?err,
-                            "Failed to replace the initial remote events in the event cache"
-                        );
-                        break;
-                    }
-                };
+                let (initial_events, recv) =
+                    match room_event_cache.subscribe_to_pinned_events().await {
+                        Ok(initial_events) => initial_events,
+                        Err(err) => {
+                            error!(
+                                ?err,
+                                "Failed to replace the initial remote events in the event cache"
+                            );
+                            break;
+                        }
+                    };
 
+                pinned_events_recv = recv;
                 timeline_controller
                     .replace_with_initial_remote_events(initial_events, RemoteEventOrigin::Cache)
                     .await;
@@ -130,7 +131,8 @@ pub(in crate::timeline) async fn event_focused_task(
                     }
                 };
 
-                let (initial_events, _) = cache.subscribe().await;
+                let (initial_events, recv) = cache.subscribe().await;
+                event_focused_events_recv = recv;
 
                 timeline_controller
                     .replace_with_initial_remote_events(initial_events, RemoteEventOrigin::Cache)

Modified file: crates/matrix-sdk/src/event_cache/caches/event_focused/mod.rs+46 additions−14 deletions@@ -46,7 +46,7 @@ use tokio::sync::{
     RwLock,
     broadcast::{Receiver, Sender},
 };
-use tracing::{instrument, trace};
+use tracing::{instrument, trace, warn};
 
 #[cfg(feature = "e2e-encryption")]
 use crate::event_cache::redecryptor::ResolvedUtd;
@@ -108,6 +108,9 @@ struct EventFocusedCacheInner {
     /// The pagination mode (room or thread).
     pagination_mode: EventFocusedPaginationMode,
 
+    /// How the cache was started, so a clear can start it again.
+    context: Option<(u16, EventFocusThreadMode)>,
+
     /// The linked chunk for this event-focused cache.
     chunk: EventLinkedChunk,
 
@@ -145,14 +148,28 @@ impl EventFocusedCacheInner {
         room: Room,
         num_context_events: u16,
         thread_mode: EventFocusThreadMode,
-    ) -> Result<StartFromResult> {
+    ) -> Result<Option<StartFromResult>> {
         trace!(num_context_events, "fetching event with context via /context");
 
+        let invalidation_lock = self.invalidation_lock.clone();
+        let generation = {
+            let _guard = invalidation_lock.lock().unwrap();
+            self.synchronize_generation();
+            self.loaded_generation
+        };
+
         let paginator = Paginator::new(room);
 
         let result =
             paginator.start_from(&self.focused_event_id, UInt::from(num_context_events)).await?;
 
+        let _invalidation_guard = invalidation_lock.lock().unwrap();
+        if generation != self.generation.load(Ordering::SeqCst) {
+            self.synchronize_generation();
+            return Ok(None);
+        }
+        self.context = Some((num_context_events, thread_mode));
+
         // Detect if the focused event is part of a thread.
         let thread_root = match thread_mode {
             EventFocusThreadMode::ForceThread => {
@@ -253,16 +270,11 @@ impl EventFocusedCacheInner {
 
         self.propagate_changes();
 
-        // Empty the updates_as_vector_diffs(), since it's impossible for an observer to
-        // have subscribed to this cache yet, since this code is part of the constructor
-        // flow.
-        //
-        // If we didn't empty those, such initial updates would be duplicated, since the
-        // subscriber would get the full initial list of events as diffs and as a set of
-        // initial events.
-        let _ = self.chunk.updates_as_vector_diffs();
+        // Draining the diffs keeps a later subscriber from seeing the initial events
+        // twice, and refills subscribers that saw a clear.
+        self.notify_subscribers(EventsOrigin::Cache);
 
-        Ok(result)
+        Ok(Some(result))
     }
 
     /// Add initial events to the chunk, with gaps for pagination tokens.
@@ -357,7 +369,7 @@ impl EventFocusedCacheInner {
         let _invalidation_guard = invalidation_lock.lock().unwrap();
         if generation != self.generation.load(Ordering::SeqCst) {
             self.synchronize_generation();
-            return Ok(PaginationResult { events: Vec::new(), hit_end_of_timeline: true });
+            return Ok(PaginationResult { events: Vec::new(), hit_end_of_timeline: false });
         }
 
         let (mut events, new_token) = pagination?;
@@ -484,7 +496,7 @@ impl EventFocusedCacheInner {
         let _invalidation_guard = invalidation_lock.lock().unwrap();
         if generation != self.generation.load(Ordering::SeqCst) {
             self.synchronize_generation();
-            return Ok(PaginationResult { events: Vec::new(), hit_end_of_timeline: true });
+            return Ok(PaginationResult { events: Vec::new(), hit_end_of_timeline: false });
         }
 
         let (events, new_token) = pagination?;
@@ -599,6 +611,7 @@ impl EventFocusedCache {
                 room,
                 focused_event_id,
                 pagination_mode: EventFocusedPaginationMode::Room { hide_thread_events: false },
+                context: None,
                 chunk: EventLinkedChunk::new(),
                 sender: sender.clone(),
                 linked_chunk_update_sender: linked_chunk_update_sender.clone(),
@@ -611,7 +624,26 @@ impl EventFocusedCache {
         }
     }
 
+    /// Empty this cache, keeping its subscribers, and start it again from its
+    /// focused event in the background.
     pub(super) fn clear(&self) {
+        self.invalidate();
+        let cache = self.clone();
+        matrix_sdk_common::executor::spawn(async move {
+            let mut inner = cache.inner.write().await;
+            let Some((num_context_events, thread_mode)) = inner.context else {
+                return;
+            };
+            let Some(room) = inner.room.get() else {
+                return;
+            };
+            if let Err(error) = inner.start_from(room, num_context_events, thread_mode).await {
+                warn!(%error, "failed to restart the event-focused cache after a clear");
+            }
+        });
+    }
+
+    pub(super) fn invalidate(&self) {
         let _guard = self.invalidation_lock.lock().unwrap();
         self.generation.fetch_add(1, Ordering::SeqCst);
         let _ = self.sender.send(TimelineVectorDiffs {
@@ -660,7 +692,7 @@ impl EventFocusedCache {
         room: Room,
         num_context_events: u16,
         thread_mode: EventFocusThreadMode,
-    ) -> Result<StartFromResult> {
+    ) -> Result<Option<StartFromResult>> {
         self.inner.write().await.start_from(room, num_context_events, thread_mode).await
     }
 

Modified file: crates/matrix-sdk/src/event_cache/caches/pinned_events/mod.rs+30 additions−14 deletions@@ -12,7 +12,11 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::{cmp::Ordering, collections::BTreeSet, sync::Arc};
+use std::{
+    cmp::Ordering,
+    collections::BTreeSet,
+    sync::{Arc, Mutex},
+};
 
 use futures_util::{StreamExt as _, stream};
 use matrix_sdk_base::{
@@ -209,9 +213,11 @@ impl PinnedEventCacheState {
 pub struct PinnedEventCache {
     state: Arc<PinnedEventCacheStateLock>,
 
+    room: WeakRoom,
+
     /// The task handling the refreshing of pinned events for this specific
     /// room.
-    _task: Arc<BackgroundTaskHandle>,
+    task: Arc<Mutex<BackgroundTaskHandle>>,
 }
 
 impl PinnedEventCache {
@@ -236,18 +242,21 @@ impl PinnedEventCache {
             store,
         };
         let state = Arc::new(PinnedEventCacheStateLock::new_inner(state));
+        let weak_room =
+            WeakRoom::new(WeakClient::from_client(&room.client()), room.room_id().to_owned());
+        let task = Arc::new(Mutex::new(Self::spawn_listener(room, state.clone())));
 
-        let task = Arc::new(
-            room.client()
-                .task_monitor()
-                .spawn_infinite_task(
-                    "pinned_event_listener_task",
-                    Self::pinned_event_listener_task(room, state.clone()),
-                )
-                .abort_on_drop(),
-        );
-
-        Self { state, _task: task }
+        Self { state, room: weak_room, task }
+    }
+
+    fn spawn_listener(room: Room, state: Arc<PinnedEventCacheStateLock>) -> BackgroundTaskHandle {
+        room.client()
+            .task_monitor()
+            .spawn_infinite_task(
+                "pinned_event_listener_task",
+                Self::pinned_event_listener_task(room, state),
+            )
+            .abort_on_drop()
     }
 
     /// Subscribe to live events from this room's pinned events cache.
@@ -260,13 +269,20 @@ impl PinnedEventCache {
         Ok((events, recv))
     }
 
+    /// Empty this cache, keeping its subscribers, and reload the pinned events
+    /// as a fresh listener would.
     pub(super) async fn clear(&self) -> Result<()> {
-        self._task.abort();
+        self.task.lock().unwrap().abort();
         let mut guard = self.state.write().await?;
         guard.state.generation += 1;
         guard.state.chunk.reset();
         guard.propagate_changes().await?;
         guard.notify_subscribers(EventsOrigin::Cache);
+        drop(guard);
+
+        if let Some(room) = self.room.get() {
+            *self.task.lock().unwrap() = Self::spawn_listener(room, self.state.clone());
+        }
 
         Ok(())
     }

Modified file: crates/matrix-sdk/src/event_cache/caches/room/mod.rs+1−1

@@ -231,7 +231,7 @@ impl RoomEventCache {

231231

232232 let mut guard = self.inner.state.write().await?;

233233 if self.inner.generation.load(Ordering::SeqCst) != generation {

234- cache.clear();

234+ cache.invalidate();

235235 continue;

236236 }

237237

Modified file: crates/matrix-sdk/src/event_cache/caches/room/state.rs+2 additions−2 deletions@@ -862,10 +862,10 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
     }
 
     pub async fn clear_loaded_views(&mut self) -> Result<(), EventCacheError> {
-        for (_, focused) in self.state.event_focused_caches.drain() {
+        for focused in self.state.event_focused_caches.values() {
             focused.clear();
         }
-        if let Some(pinned) = self.state.pinned_event_cache.take() {
+        if let Some(pinned) = self.state.pinned_event_cache.get() {
             pinned.clear().await?;
         }
         for thread in self.state.threads.values_mut() {

Modified file: crates/matrix-sdk/tests/room_clear.rs+48 additions−9 deletions@@ -12,7 +12,7 @@ use matrix_sdk::{
 use matrix_sdk_test::{ALICE, JoinedRoomBuilder, event_factory::EventFactory};
 
 #[tokio::test]
-async fn clear_invalidates_focused_cache_clones_and_inflight_pagination() {
+async fn clear_restarts_focused_cache_and_fences_inflight_pagination() {
     let server = MatrixMockServer::new().await;
     let client = server.client_builder().build().await;
     let room_id = room_id!("!clear:localhost");
@@ -76,17 +76,45 @@ async fn clear_invalidates_focused_cache_clones_and_inflight_pagination() {
     .await
     .unwrap();
 
+    server
+        .mock_room_event_context()
+        .ok(RoomContextResponseTemplate::new(
+            factory.text_msg("focused").event_id(focused_id).into_event(),
+        )
+        .start("fresh-focused-cursor"))
+        .mock_once()
+        .mount()
+        .await;
     tokio::time::timeout(Duration::from_secs(2), cache.clear()).await.unwrap().unwrap();
     let clear =
         tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
     assert!(matches!(clear.diffs.as_slice(), [VectorDiff::Clear]));
-    tokio::time::timeout(Duration::from_secs(3), pagination).await.unwrap().unwrap().unwrap();
+    let interrupted =
+        tokio::time::timeout(Duration::from_secs(3), pagination).await.unwrap().unwrap().unwrap();
+    assert!(interrupted.events.is_empty());
+    assert!(!interrupted.hit_end_of_timeline);
+
+    let mut visible = imbl::Vector::new();
+    tokio::time::timeout(Duration::from_secs(3), async {
+        while visible.is_empty() {
+            for diff in updates.recv().await.unwrap().diffs {
+                diff.apply(&mut visible);
+            }
+        }
+    })
+    .await
+    .unwrap();
+    let ids: Vec<_> = visible.iter().filter_map(|event| event.event_id()).collect();
+    assert_eq!(ids, vec![focused_id]);
     let (cloned_events, _cloned_updates) = focused_clone.subscribe().await;
-    assert!(cloned_events.is_empty());
-    let result = focused_clone.paginate_backwards(10).await.unwrap();
-    assert!(result.hit_end_of_timeline);
-    let (events_after_pagination, _updates_after_pagination) = focused_clone.subscribe().await;
-    assert!(events_after_pagination.is_empty());
+    assert_eq!(cloned_events.len(), 1);
+    assert!(
+        cache
+            .get_event_focused_cache(focused_id.to_owned(), EventFocusThreadMode::Automatic)
+            .await
+            .unwrap()
+            .is_some()
+    );
 }
 
 #[tokio::test]
@@ -131,13 +159,13 @@ async fn clear_empties_subscribed_thread_cache() {
 }
 
 #[tokio::test]
-async fn clear_empties_loaded_pinned_cache() {
+async fn clear_reloads_loaded_pinned_cache() {
     let server = MatrixMockServer::new().await;
     let room_id = room_id!("!clear:localhost");
     let pinned_event_id = event_id!("$pinned:localhost");
     let factory = EventFactory::new().room(room_id).sender(&ALICE);
     let pinned_event = factory.text_msg("pinned").event_id(pinned_event_id).into_event();
-    server.mock_room_event().match_event_id().ok(pinned_event).mock_once().mount().await;
+    server.mock_room_event().match_event_id().ok(pinned_event).mount().await;
     let client = server.client_builder().build().await;
     client.event_cache().subscribe().unwrap();
     let room = server
@@ -171,4 +199,15 @@ async fn clear_empties_loaded_pinned_cache() {
         diff.apply(&mut visible);
     }
     assert!(visible.is_empty());
+    tokio::time::timeout(Duration::from_secs(3), async {
+        while visible.is_empty() {
+            for diff in updates.recv().await.unwrap().diffs {
+                diff.apply(&mut visible);
+            }
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(visible.len(), 1);
+    assert_eq!(visible[0].event_id().as_deref(), Some(pinned_event_id));
 }

matrix-rust-sdk

Commit 3c85630430e1

Commit `3c85630430e1`