matrix-rust-sdk

public

Star0

Fork 0

Watch0

61 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `b9c68844ca91`

tarball

atagen <boss@atagen.co>2026-10-28 07:47 UTC

3 files changed+102-15

event-cache: hold off rooms nobody views while a room restores or fills

A room opened during a big catch-up waited 2.9s on its restore, its IndexedDB reads queued behind four other rooms' cache creation and sync handling at a time, 470 rooms of it. A restore, recent load or gap fill now keeps sync updates for unviewed rooms waiting until it finishes, for 3s at most per stretch, while rooms on screen still go first.

Modified file: crates/matrix-sdk/src/event_cache/caches/room/history.rs+7 additions@@ -150,6 +150,13 @@ async fn coordinate_history(
         };
         let cache = RoomEventCache { inner };
         let recent = matches!(operation, HistoryOperation::Recent);
+        let _background = matches!(
+            operation,
+            HistoryOperation::Restore | HistoryOperation::Recent | HistoryOperation::Fill(_)
+        )
+        .then(|| cache.inner.weak_room.get())
+        .flatten()
+        .map(|room| room.client().event_cache().hold_background());
         let name = operation.name();
         tracing::debug!(room_id = %cache.room_id(), operation = name, "history operation started");
         let mut waiters = vec![reply];

Modified file: crates/matrix-sdk/src/event_cache/mod.rs+56 additions−13 deletions@@ -48,7 +48,7 @@ use matrix_sdk_base::{
     timer,
 };
 use matrix_sdk_common::timeout::timeout;
-use ruma::{OwnedRoomId, RoomId};
+use ruma::{OwnedRoomId, RoomId, time::Instant};
 use tokio::sync::{
     Mutex, RwLock,
     broadcast::{Receiver, Sender, channel},
@@ -248,6 +248,7 @@ impl EventCache {
                 clears: Default::default(),
                 dropped_room_updates: Default::default(),
                 room_updates_handled: watch::Sender::new(0),
+                background_holds: Default::default(),
                 by_room: Default::default(),
                 drop_handles: Default::default(),
                 auto_shrink_sender: Default::default(),
@@ -395,6 +396,17 @@ impl EventCache {
         self.inner.dropped_room_updates.load(Ordering::SeqCst)
     }
 
+    /// Holds off sync updates for rooms nobody views until dropped, for
+    /// [`BACKGROUND_HOLD`] at most, so a room's history work isn't queued
+    /// behind a catch-up's storage traffic.
+    pub(crate) fn hold_background(&self) -> BackgroundHold {
+        self.inner.background_holds.send_modify(|holds| {
+            holds.count += 1;
+            holds.since.get_or_insert_with(Instant::now);
+        });
+        BackgroundHold { inner: self.inner.clone() }
+    }
+
     /// Check whether [`EventCache::subscribe`] has been called.
     pub fn has_subscribed(&self) -> bool {
         self.inner.drop_handles.get().is_some()
@@ -539,6 +551,9 @@ struct EventCacheInner {
     /// the client's `room_updates_sent` terms.
     room_updates_handled: watch::Sender<u64>,
 
+    /// History work in progress, holding off rooms nobody views.
+    background_holds: watch::Sender<BackgroundHolds>,
+
     /// Lazily-filled cache of live [`RoomEventCache`], once per room.
     //
     // It's behind an `Arc` to get owned locks.
@@ -604,6 +619,34 @@ const ROOM_UPDATES_AT_ONCE: usize = 1;
 #[cfg(target_family = "wasm")]
 const ROOM_UPDATES_AT_ONCE: usize = 4;
 
+/// How long a room's history work can hold off sync updates for rooms nobody
+/// views. Opening a room during a 470 room catch-up otherwise queued its
+/// restore's IndexedDB reads behind every other room's, for 2.9s.
+const BACKGROUND_HOLD: Duration = Duration::from_secs(3);
+
+#[derive(Debug, Default)]
+struct BackgroundHolds {
+    count: usize,
+    since: Option<Instant>,
+}
+
+/// Holds off sync updates for rooms nobody views while alive, see
+/// [`EventCache::hold_background`].
+pub(crate) struct BackgroundHold {
+    inner: Arc<EventCacheInner>,
+}
+
+impl Drop for BackgroundHold {
+    fn drop(&mut self) {
+        self.inner.background_holds.send_modify(|holds| {
+            holds.count -= 1;
+            if holds.count == 0 {
+                holds.since = None;
+            }
+        });
+    }
+}
+
 /// One room's part of a [`RoomUpdates`].
 enum RoomUpdate {
     Left(LeftRoomUpdate),
@@ -736,19 +779,19 @@ impl EventCacheInner {
 
     /// Handle the next few rooms of `pending`, those on screen first, with the
     /// `multiple_room_updates_lock` held.
+    async fn viewed_pending(&self, pending: &PendingRoomUpdates) -> 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()
+    }
+
     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()
-        };
+        let viewed = self.viewed_pending(pending).await;
         // 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()

Modified file: crates/matrix-sdk/src/event_cache/tasks.rs+39 additions−2 deletions@@ -20,7 +20,7 @@ use std::{
 use eyeball::Subscriber;
 use matrix_sdk_base::{
     linked_chunk::OwnedLinkedChunkId, serde_helpers::extract_thread_root_from_content,
-    sync::RoomUpdates,
+    sleep::sleep, sync::RoomUpdates,
 };
 use ruma::{OwnedEventId, OwnedTransactionId};
 use tokio::{
@@ -36,7 +36,7 @@ use tokio::{
 use tracing::{Instrument as _, Span, debug, error, info, info_span, instrument, trace, warn};
 
 use super::{
-    AutoShrinkChannelPayload, EventCacheInner, EventsOrigin, PendingRoomUpdates,
+    AutoShrinkChannelPayload, BACKGROUND_HOLD, EventCacheInner, EventsOrigin, PendingRoomUpdates,
     RoomEventCacheLinkedChunkUpdate, RoomEventCacheUpdate, TimelineVectorDiffs,
 };
 use crate::{
@@ -78,6 +78,10 @@ pub(super) async fn room_updates_task(
             }
         }
 
+        if inner.viewed_pending(&pending).await.is_empty() {
+            wait_out_background_hold(&inner, &mut room_updates_feed, &mut pending).await;
+        }
+
         {
             let _lock = inner.multiple_room_updates_lock.lock().await;
             inner.handle_pending_room_updates(&mut pending).await;
@@ -102,6 +106,39 @@ pub(super) async fn room_updates_task(
     }
 }
 
+/// Wait while a room's history work holds off rooms nobody views, until it
+/// ends, [`BACKGROUND_HOLD`] passes, or an update for a viewed room arrives.
+async fn wait_out_background_hold(
+    inner: &EventCacheInner,
+    feed: &mut Receiver<RoomUpdates>,
+    pending: &mut PendingRoomUpdates,
+) {
+    let mut holds = inner.background_holds.subscribe();
+    loop {
+        let since = holds.borrow_and_update().since;
+        let Some(left) = since.and_then(|since| BACKGROUND_HOLD.checked_sub(since.elapsed()))
+        else {
+            return;
+        };
+        select! {
+            changed = holds.changed() => if changed.is_err() {
+                return;
+            },
+            () = sleep(left) => return,
+            update = feed.recv() => {
+                match update {
+                    Ok(updates) => pending.fold(updates),
+                    Err(RecvError::Lagged(num_skipped)) => pending.lagged += num_skipped,
+                    Err(RecvError::Closed) => return,
+                }
+                if !inner.viewed_pending(pending).await.is_empty() {
+                    return;
+                }
+            }
+        }
+    }
+}
+
 /// Listen to _ignore user list update changes_ to clear the rooms when a user
 /// is ignored or unignored.
 #[instrument(skip_all)]

matrix-rust-sdk

Commit b9c68844ca91

Commit `b9c68844ca91`