matrix-rust-sdk

public

Star0

Fork 0

Watch0

59 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `8d754af8d109`

tarball

atagen <boss@atagen.co>2026-10-28 05:03 UTC

6 files changed+201-83

event-cache: fold queued sync responses together, and handle rooms on screen first

The room updates task handled one response at a time, rooms in id order, so during a catch-up the room someone had open waited behind every queued response, and live messages did not show until the backlog drained. It now takes every queued response, keeps each room's updates in the order sync sent them, handles rooms with a live `Timeline` first through the new `RoomEventCache::view` marker, and on wasm handles four rooms at once to overlap their IndexedDB waits. A response only counts as handled once every room from the responses folded with it is done, so the saved pos still never runs ahead of the store. The search indexer subscribes to nearly every active room, which is why priority uses the marker rather than the subscriber count.

Modified file: crates/matrix-sdk-ui/src/timeline/builder.rs+2 additions@@ -256,6 +256,7 @@ impl TimelineBuilder {
 
         let crypto_drop_handles = spawn_crypto_tasks(controller.clone()).await;
 
+        let view = room_event_cache.view();
         let timeline = Timeline {
             controller,
             event_cache: room_event_cache,
@@ -266,6 +267,7 @@ impl TimelineBuilder {
                 _local_echo_listener_handle: local_echo_listener_handle,
                 _focus_drop_handle: focus_task,
                 _event_cache_drop_handle: event_cache_drop,
+                _view: view,
             }),
         };
 

Modified file: crates/matrix-sdk-ui/src/timeline/mod.rs+4 additions−1 deletions@@ -29,7 +29,9 @@ use matrix_sdk::{
     Result,
     attachment::{AttachmentInfo, Thumbnail},
     deserialized_responses::TimelineEvent,
-    event_cache::{EventCacheDropHandles, EventFocusThreadMode, RoomEventCache},
+    event_cache::{
+        EventCacheDropHandles, EventFocusThreadMode, RoomEventCache, RoomEventCacheView,
+    },
     room::{
         Receipts, Room,
         edit::EditedContent,
@@ -932,6 +934,7 @@ struct TimelineDropHandle {
     _event_cache_drop_handle: Arc<EventCacheDropHandles>,
     _focus_drop_handle: Option<BackgroundTaskHandle>,
     _crypto_drop_handles: CryptoDropHandles,
+    _view: RoomEventCacheView,
 }
 
 #[cfg(not(target_family = "wasm"))]

Modified file: crates/matrix-sdk-ui/tests/integration/room_list_service.rs+14 additions−14 deletions@@ -1956,16 +1956,6 @@ async fn test_room_sorting() -> Result<(), Error> {
         end;
     };
 
-    // All rooms get new messages, so their entries will get updates because of read
-    // receipt updates.
-    //
-    // Starting with r0.
-    assert_entries_batch! {
-        [stream]
-        set [ 0 ] [ "!r0:bar.org" ];
-        end;
-    };
-
     // Now we have:
     //
     // | index | room ID | recency | name |
@@ -1983,10 +1973,13 @@ async fn test_room_sorting() -> Result<(), Error> {
         end;
     };
 
-    // Read receipt update for r1.
+    // All rooms get new messages, so their entries will get updates because of read
+    // receipt updates, each landing once the next room has moved.
+    //
+    // Starting with r0.
     assert_entries_batch! {
         [stream]
-        set [ 1 ] [ "!r1:bar.org" ];
+        set [ 0 ] [ "!r0:bar.org" ];
         end;
     };
 
@@ -2007,6 +2000,13 @@ async fn test_room_sorting() -> Result<(), Error> {
         end;
     };
 
+    // Read receipt update for r1.
+    assert_entries_batch! {
+        [stream]
+        set [ 2 ] [ "!r1:bar.org" ];
+        end;
+    };
+
     // Read receipt update for r2.
     assert_entries_batch! {
         [stream]
@@ -2126,12 +2126,12 @@ async fn test_room_sorting() -> Result<(), Error> {
     // Rooms are individually updated.
     assert_entries_batch! {
         [stream]
-        set [ 0 ] [ "!r3:bar.org" ];
+        set [ 2 ] [ "!r6:bar.org" ];
         end;
     };
     assert_entries_batch! {
         [stream]
-        set [ 2 ] [ "!r6:bar.org" ];
+        set [ 0 ] [ "!r3:bar.org" ];
         end;
     };
     assert_entries_batch! {

Modified file: crates/matrix-sdk/src/event_cache/caches/room/mod.rs+28 additions−1 deletions@@ -26,7 +26,7 @@ use std::{
     fmt,
     sync::{
         Arc,
-        atomic::{AtomicU64, Ordering},
+        atomic::{AtomicU64, AtomicUsize, Ordering},
     },
 };
 
@@ -119,6 +119,17 @@ impl RoomEventCache {
         Ok(state.room_linked_chunk().events().map(|(_position, item)| item.clone()).collect())
     }
 
+    /// Mark this room as on screen for as long as the returned value lives, so
+    /// sync handling takes its updates ahead of other rooms'.
+    pub fn view(&self) -> RoomEventCacheView {
+        self.inner.viewers.fetch_add(1, Ordering::SeqCst);
+        RoomEventCacheView { viewers: self.inner.viewers.clone() }
+    }
+
+    pub(in super::super) fn is_viewed(&self) -> bool {
+        self.inner.viewers.load(Ordering::SeqCst) > 0
+    }
+
     /// Subscribe to this room updates, after getting the initial list of
     /// events.
     ///
@@ -458,6 +469,18 @@ impl RoomEventCache {
     }
 }
 
+/// Keeps a room ahead in sync handling while it lives, see [`RoomEventCache::view`].
+#[derive(Debug)]
+pub struct RoomEventCacheView {
+    viewers: Arc<AtomicUsize>,
+}
+
+impl Drop for RoomEventCacheView {
+    fn drop(&mut self) {
+        self.viewers.fetch_sub(1, Ordering::SeqCst);
+    }
+}
+
 /// The (non-cloneable) details of the `RoomEventCache`.
 pub(super) struct RoomEventCacheInner {
     history_coordinator: std::sync::OnceLock<mpsc::Sender<history::HistoryRequest>>,
@@ -483,6 +506,9 @@ pub(super) struct RoomEventCacheInner {
 
     /// Update sender for this room.
     update_sender: RoomEventCacheUpdateSender,
+
+    /// How many [`RoomEventCacheView`]s are alive.
+    viewers: Arc<AtomicUsize>,
 }
 
 impl RoomEventCacheInner {
@@ -507,6 +533,7 @@ impl RoomEventCacheInner {
             pagination_batch_token_notifier: Default::default(),
             auto_shrink_sender,
             shared_pagination_status,
+            viewers: Default::default(),
         }
     }
 

Modified file: crates/matrix-sdk/src/event_cache/mod.rs+112 additions−38 deletions@@ -28,7 +28,7 @@
 #![forbid(missing_docs)]
 
 use std::{
-    collections::HashMap,
+    collections::{BTreeMap, HashMap},
     fmt,
     sync::{
         Arc, Mutex as StdMutex, OnceLock, PoisonError, RwLock as StdRwLock, RwLockReadGuard,
@@ -38,12 +38,12 @@ use std::{
     time::Duration,
 };
 
-use futures_util::future::try_join_all;
+use futures_util::{StreamExt, future::try_join_all};
 use matrix_sdk_base::{
     cross_process_lock::CrossProcessLockError,
     event_cache::store::{EventCacheStoreError, EventCacheStoreLock, EventCacheStoreLockState},
     linked_chunk::lazy_loader::LazyLoaderError,
-    sync::RoomUpdates,
+    sync::{JoinedRoomUpdate, LeftRoomUpdate, RoomUpdates},
     task_monitor::BackgroundTaskHandle,
     timer,
 };
@@ -78,7 +78,7 @@ pub use caches::{
     room::{
         HistoryGapFillOutcome, HistoryGapPosition, RecentHistoryContinuity, RecentHistoryOutcome,
         RoomEventCache, RoomEventCacheGenericUpdate, RoomEventCacheSubscriber,
-        RoomEventCacheUpdate, pagination::RoomPagination,
+        RoomEventCacheUpdate, RoomEventCacheView, pagination::RoomPagination,
     },
     thread::pagination::ThreadPagination,
 };
@@ -599,6 +599,60 @@ struct EventCacheInner {
 
 type AutoShrinkChannelPayload = OwnedRoomId;
 
+// NOTE: We tried to make this concurrent at some point, but it turned out to be
+// a performance regression, even for large sync updates. Lacking time
+// to investigate, this code remains sequential for now. See also
+// https://github.com/matrix-org/matrix-rust-sdk/pull/5426.
+#[cfg(not(target_family = "wasm"))]
+const ROOM_UPDATES_AT_ONCE: usize = 1;
+
+/// How many rooms' sync updates are handled at once. On wasm a room's handling is
+/// mostly waiting on IndexedDB, which other rooms can use.
+#[cfg(target_family = "wasm")]
+const ROOM_UPDATES_AT_ONCE: usize = 4;
+
+/// One room's part of a [`RoomUpdates`].
+enum RoomUpdate {
+    Left(LeftRoomUpdate),
+    Joined(JoinedRoomUpdate),
+}
+
+/// Sync updates taken off the broadcast and not handled yet, per room, in the
+/// order sync sent them.
+#[derive(Default)]
+struct PendingRoomUpdates {
+    rooms: BTreeMap<OwnedRoomId, Vec<RoomUpdate>>,
+
+    /// Broadcast messages folded in since `rooms` was last empty.
+    folded: u64,
+
+    /// Broadcast messages lost to a lag since `rooms` was last empty.
+    lagged: u64,
+}
+
+impl PendingRoomUpdates {
+    fn fold(&mut self, updates: RoomUpdates) {
+        self.folded += 1;
+        for (room_id, update) in updates.left {
+            self.rooms.entry(room_id).or_default().push(RoomUpdate::Left(update));
+        }
+        for (room_id, update) in updates.joined {
+            self.rooms.entry(room_id).or_default().push(RoomUpdate::Joined(update));
+        }
+        // Invited rooms.
+        // TODO: we don't anything with `updates.invite` at this point.
+    }
+
+    fn is_empty(&self) -> bool {
+        self.rooms.is_empty()
+    }
+
+    /// How many broadcast messages are fully handled, once `rooms` is empty.
+    fn take_handled(&mut self) -> u64 {
+        std::mem::take(&mut self.folded) + std::mem::take(&mut self.lagged)
+    }
+}
+
 impl EventCacheInner {
     fn client(&self) -> Result<Client> {
         self.client.get().ok_or(EventCacheError::ClientDropped)
@@ -678,49 +732,69 @@ impl EventCacheInner {
             self.multiple_room_updates_lock.lock().await
         };
 
-        // NOTE: We tried to make this concurrent at some point, but it turned out to be
-        // a performance regression, even for large sync updates. Lacking time
-        // to investigate, this code remains sequential for now. See also
-        // https://github.com/matrix-org/matrix-rust-sdk/pull/5426.
+        let mut pending = PendingRoomUpdates::default();
+        pending.fold(updates);
+        while !pending.is_empty() {
+            self.handle_pending_room_updates(&mut pending).await;
+        }
 
-        // Left rooms.
-        for (room_id, left_room_update) in updates.left {
-            let caches = match self.all_caches_for_room(&room_id).await {
-                Ok(caches) => caches,
-                Err(err) => {
-                    self.drop_room_update(&room_id, err);
-                    continue;
-                }
-            };
+        Ok(())
+    }
 
-            if let Err(err) = caches.handle_left_room_update(left_room_update).await {
-                // Non-fatal error, try to continue to the next room.
-                error!("handling left room update: {err}");
-            }
-        }
+    /// Handle the next few rooms of `pending`, those on screen first, with the
+    /// `multiple_room_updates_lock` held.
+    async fn handle_pending_room_updates(&self, pending: &mut PendingRoomUpdates) {
+        let viewed: Vec<OwnedRoomId> = {
+            let by_room = self.by_room.read().await;
+            pending
+                .rooms
+                .keys()
+                .filter(|room_id| {
+                    by_room.get(*room_id).is_some_and(|caches| caches.room.is_viewed())
+                })
+                .take(ROOM_UPDATES_AT_ONCE)
+                .cloned()
+                .collect()
+        };
+        // rooms on screen get a round of their own, so a slow cold room can't hold them back
+        let round: Vec<OwnedRoomId> = if viewed.is_empty() {
+            pending.rooms.keys().take(ROOM_UPDATES_AT_ONCE).cloned().collect()
+        } else {
+            viewed
+        };
+        let round: Vec<_> =
+            round.into_iter().filter_map(|room_id| pending.rooms.remove_entry(&room_id)).collect();
 
-        // Joined rooms.
-        for (room_id, joined_room_update) in updates.joined {
-            trace!(?room_id, "Handling a `JoinedRoomUpdate`");
+        futures_util::stream::iter(round)
+            .for_each_concurrent(ROOM_UPDATES_AT_ONCE, |(room_id, updates)| async move {
+                self.handle_room_queue(&room_id, updates).await;
+            })
+            .await;
+    }
 
-            let caches = match self.all_caches_for_room(&room_id).await {
-                Ok(caches) => caches,
-                Err(err) => {
-                    self.drop_room_update(&room_id, err);
-                    continue;
+    /// Handle one room's updates, in the order sync sent them.
+    async fn handle_room_queue(&self, room_id: &RoomId, updates: Vec<RoomUpdate>) {
+        let caches = match self.all_caches_for_room(room_id).await {
+            Ok(caches) => caches,
+            Err(err) => {
+                self.drop_room_update(room_id, err);
+                return;
+            }
+        };
+
+        for update in updates {
+            let handled = match update {
+                RoomUpdate::Left(update) => caches.handle_left_room_update(update).await,
+                RoomUpdate::Joined(update) => {
+                    trace!(?room_id, "Handling a `JoinedRoomUpdate`");
+                    caches.handle_joined_room_update(update).await
                 }
             };
-
-            if let Err(err) = caches.handle_joined_room_update(joined_room_update).await {
+            if let Err(err) = handled {
                 // Non-fatal error, try to continue to the next room.
-                error!(%room_id, "handling joined room update: {err}");
+                error!(%room_id, "handling a room update: {err}");
             }
         }
-
-        // Invited rooms.
-        // TODO: we don't anything with `updates.invite` at this point.
-
-        Ok(())
     }
 
     /// Return all the event caches associated to a specific room.

Modified file: crates/matrix-sdk/src/event_cache/tasks.rs+41 additions−29 deletions@@ -26,14 +26,17 @@ use ruma::{OwnedEventId, OwnedTransactionId};
 use tokio::{
     select,
     sync::{
-        broadcast::{Receiver, Sender, error::RecvError},
+        broadcast::{
+            Receiver, Sender,
+            error::{RecvError, TryRecvError},
+        },
         mpsc,
     },
 };
 use tracing::{Instrument as _, Span, debug, error, info, info_span, instrument, trace, warn};
 
 use super::{
-    AutoShrinkChannelPayload, EventCacheError, EventCacheInner, EventsOrigin,
+    AutoShrinkChannelPayload, EventCacheInner, EventsOrigin, PendingRoomUpdates,
     RoomEventCacheLinkedChunkUpdate, RoomEventCacheUpdate, TimelineVectorDiffs,
 };
 use crate::{
@@ -48,44 +51,53 @@ pub(super) async fn room_updates_task(
     mut room_updates_feed: Receiver<RoomUpdates>,
 ) {
     trace!("Spawning the listen task");
+    let mut pending = PendingRoomUpdates::default();
     loop {
-        match room_updates_feed.recv().await {
-            Ok(updates) => {
-                trace!("Receiving `RoomUpdates`");
-
-                if let Err(err) = inner.handle_room_updates(updates).await {
-                    match err {
-                        EventCacheError::ClientDropped => {
-                            // The client has dropped, exit the listen task.
-                            info!(
-                                "Closing the event cache global listen task because client dropped"
-                            );
-                            break;
-                        }
-                        err => {
-                            error!("Error when handling room updates: {err}");
-                        }
-                    }
+        if pending.is_empty() {
+            match room_updates_feed.recv().await {
+                Ok(updates) => {
+                    trace!("Receiving `RoomUpdates`");
+                    pending.fold(updates);
+                }
+                Err(RecvError::Lagged(num_skipped)) => pending.lagged += num_skipped,
+                Err(RecvError::Closed) => {
+                    // The sender has shut down, exit.
+                    info!("Closing the event cache global listen task because receiver closed");
+                    break;
                 }
-                inner.room_updates_handled.send_modify(|handled| *handled += 1);
             }
+        }
 
-            Err(RecvError::Lagged(num_skipped)) => {
+        // a backlog of responses folded in together lets a room on screen skip the queue,
+        // and the room's own updates stay in the order sync sent them
+        loop {
+            match room_updates_feed.try_recv() {
+                Ok(updates) => pending.fold(updates),
+                Err(TryRecvError::Lagged(num_skipped)) => pending.lagged += num_skipped,
+                Err(TryRecvError::Empty | TryRecvError::Closed) => break,
+            }
+        }
+
+        {
+            let _lock = inner.multiple_room_updates_lock.lock().await;
+            inner.handle_pending_room_updates(&mut pending).await;
+        }
+
+        if pending.is_empty() {
+            if pending.lagged > 0 {
                 // Forget everything we know; we could have missed events, and we have
                 // no way to reconcile at the moment!
                 // TODO: implement Smart Matching™,
-                warn!(num_skipped, "Lagged behind room updates, clearing all rooms");
+                warn!(
+                    num_skipped = pending.lagged,
+                    "Lagged behind room updates, clearing all rooms"
+                );
                 if let Err(err) = inner.clear_all_rooms().await {
                     error!("when clearing storage after lag in listen_task: {err}");
                 }
-                inner.room_updates_handled.send_modify(|handled| *handled += num_skipped);
-            }
-
-            Err(RecvError::Closed) => {
-                // The sender has shut down, exit.
-                info!("Closing the event cache global listen task because receiver closed");
-                break;
             }
+            let handled = pending.take_handled();
+            inner.room_updates_handled.send_modify(|count| *count += handled);
         }
     }
 }

matrix-rust-sdk

Commit 8d754af8d109

Commit `8d754af8d109`