matrix-rust-sdk

public

Star0

Fork 0

Watch0

54 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `e4a8dbca1ac5`

tarball

atagen <boss@atagen.co>2026-10-28 01:53 UTC

3 files changed+169-52

event-cache: create a room's caches outside the room caches lock, and give up on a stuck one

On wasm one room's cache creation can stall for good, and it ran under the global room caches write lock, so every room open and every later sync waited on it until a reload. Creation now runs under a per-room gate with the lock taken only to insert, gives up on an attempt after 10s with a warning naming the step it was on, and tries once more. A room update dropped because creation failed twice stops later responses from saving pos until a restart, since that update was never stored. A clear of every room that lands mid-creation discards the stale caches and creates them again.

Modified file: crates/matrix-sdk/src/event_cache/caches/mod.rs+20 additions−1 deletions@@ -14,7 +14,7 @@
 
 use std::{
     future::Future,
-    sync::{Arc, atomic::AtomicU64},
+    sync::{Arc, Mutex as StdMutex, PoisonError, atomic::AtomicU64},
     time::Duration,
 };
 
@@ -103,6 +103,21 @@ where
     }
 }
 
+/// The step a room's cache creation is on, so a creation that never finishes says where.
+#[derive(Default)]
+pub(super) struct CreationStep(StdMutex<&'static str>);
+
+impl CreationStep {
+    pub(super) fn at(&self, room_id: &RoomId, step: &'static str) {
+        debug!(%room_id, step, "room caches creation step");
+        *self.0.lock().unwrap_or_else(PoisonError::into_inner) = step;
+    }
+
+    pub(super) fn current(&self) -> &'static str {
+        *self.0.lock().unwrap_or_else(PoisonError::into_inner)
+    }
+}
+
 /// A type to hold all the caches for a given room.
 #[derive(Debug)]
 pub(super) struct Caches {
@@ -111,6 +126,7 @@ pub(super) struct Caches {
 
 impl Caches {
     /// Create a new [`Caches`].
+    #[allow(clippy::too_many_arguments)]
     pub async fn new(
         weak_client: &WeakClient,
         room_id: &RoomId,
@@ -119,6 +135,7 @@ impl Caches {
         auto_shrink_sender: mpsc::Sender<OwnedRoomId>,
         store: EventCacheStoreLock,
         automatic_pagination: Option<AutomaticPagination>,
+        step: &CreationStep,
     ) -> Result<Self> {
         let Some(client) = weak_client.get() else {
             return Err(EventCacheError::ClientDropped);
@@ -156,9 +173,11 @@ impl Caches {
             pagination_status.clone(),
             generation.clone(),
             automatic_pagination,
+            step,
         )
         .await?;
 
+        step.at(room_id, "reading the loaded state");
         let timeline_is_not_empty =
             room_state.read().await?.room_linked_chunk().revents().next().is_some();
 

Modified file: crates/matrix-sdk/src/event_cache/caches/room/state.rs+6 additions@@ -323,7 +323,9 @@ impl LockedRoomEventCacheState {
         pagination_status: SharedObservable<SharedPaginationStatus>,
         generation: Arc<AtomicU64>,
         automatic_pagination: Option<AutomaticPagination>,
+        step: &super::super::CreationStep,
     ) -> Result<Self, EventCacheError> {
+        step.at(&room_id, "taking the store lock");
         let store_guard = match store.lock().await? {
             // Lock is clean: all good!
             EventCacheStoreLockState::Clean(guard) => guard,
@@ -345,6 +347,7 @@ impl LockedRoomEventCacheState {
         // indicates that at some point, there's some malformed data.
         // a storage error can be transient (an IndexedDB transaction aborted under us), and the
         // clear below throws away the room's whole history, so read once more before trusting it
+        step.at(&room_id, "loading the linked chunk metadata");
         let metadata = match super::super::read_store("linked chunk metadata", || {
             load_linked_chunk_metadata(&store_guard, linked_chunk_id)
         })
@@ -362,6 +365,7 @@ impl LockedRoomEventCacheState {
                 error!("error when loading a linked chunk's metadata from the store: {err}");
 
                 // Try to clear storage for this room.
+                step.at(&room_id, "clearing a malformed linked chunk");
                 store_guard
                     .handle_linked_chunk_updates(linked_chunk_id, vec![Update::Clear])
                     .await?;
@@ -371,6 +375,7 @@ impl LockedRoomEventCacheState {
             }
         };
 
+        step.at(&room_id, "loading the latest chunk");
         let last_chunk = match super::super::read_store("latest chunk", || {
             store_guard.load_last_chunk(linked_chunk_id)
         })
@@ -393,6 +398,7 @@ impl LockedRoomEventCacheState {
                 error!("error when loading a linked chunk's latest chunk from the store: {err}");
 
                 // Try to clear storage for this room.
+                step.at(&room_id, "clearing an unreadable linked chunk");
                 store_guard
                     .handle_linked_chunk_updates(linked_chunk_id, vec![Update::Clear])
                     .await?;

Modified file: crates/matrix-sdk/src/event_cache/mod.rs+143 additions−51 deletions@@ -30,7 +30,11 @@
 use std::{
     collections::HashMap,
     fmt,
-    sync::{Arc, OnceLock, RwLock as StdRwLock, RwLockReadGuard, RwLockWriteGuard},
+    sync::{
+        Arc, Mutex as StdMutex, OnceLock, PoisonError, RwLock as StdRwLock, RwLockReadGuard,
+        RwLockWriteGuard,
+        atomic::{AtomicBool, AtomicU64, Ordering},
+    },
     time::Duration,
 };
 
@@ -50,7 +54,7 @@ use tokio::sync::{
     broadcast::{Receiver, Sender, channel},
     mpsc, watch,
 };
-use tracing::{debug, error, instrument, trace};
+use tracing::{debug, error, instrument, trace, warn};
 
 use crate::{
     Client,
@@ -148,8 +152,20 @@ pub enum EventCacheError {
     /// was cancelled by a runtime shutdown.
     #[error("the event cache task did not complete: {0}")]
     TaskInterrupted(String),
+
+    /// A room's caches were not created in time, twice.
+    #[error("the caches of room {room_id} were not created in time, stuck {step}")]
+    RoomCachesNotCreated {
+        /// The room whose caches were not created.
+        room_id: OwnedRoomId,
+        /// The creation step the last attempt was on.
+        step: &'static str,
+    },
 }
 
+/// How long creating one room's caches may take before the attempt is abandoned.
+const CREATION_WAIT: Duration = Duration::from_secs(10);
+
 /// A result using the [`EventCacheError`].
 pub type Result<T> = std::result::Result<T, EventCacheError>;
 
@@ -228,6 +244,9 @@ impl EventCache {
                 config: StdRwLock::new(EventCacheConfig::default()),
                 store: event_cache_store,
                 multiple_room_updates_lock: Default::default(),
+                creation_gates: Default::default(),
+                clears: Default::default(),
+                dropped_room_updates: Default::default(),
                 room_updates_handled: watch::Sender::new(0),
                 by_room: Default::default(),
                 drop_handles: Default::default(),
@@ -374,6 +393,9 @@ impl EventCache {
         if !self.has_subscribed() {
             return true;
         }
+        if self.inner.dropped_room_updates.load(Ordering::SeqCst) {
+            return false;
+        }
         let target = client.room_updates_sent();
         let mut handled = self.inner.room_updates_handled.subscribe();
         let waited = timeout(handled.wait_for(|handled| *handled >= target), limit).await;
@@ -510,6 +532,16 @@ struct EventCacheInner {
     // TODO: that's the place to add a cross-process lock!
     multiple_room_updates_lock: Mutex<()>,
 
+    /// One creation at a time per room, see [`EventCacheInner::all_caches_for_room`].
+    creation_gates: StdMutex<HashMap<OwnedRoomId, Arc<Mutex<()>>>>,
+
+    /// How many times every room has been cleared.
+    clears: AtomicU64,
+
+    /// Whether a room update was dropped because the room's caches couldn't be
+    /// created, after which nothing counts as handled until a restart.
+    dropped_room_updates: AtomicBool,
+
     /// How many of the client's [`RoomUpdates`] have been handled, counted in
     /// the client's `room_updates_sent` terms.
     room_updates_handled: watch::Sender<u64>,
@@ -615,6 +647,7 @@ impl EventCacheInner {
         //    will take turn to acquire it.
 
         let mut all_caches = self.by_room.write().await;
+        self.clears.fetch_add(1, Ordering::SeqCst);
 
         // Prepare to reset all the caches: it ensures nobody is accessing or mutating
         // them.
@@ -652,9 +685,12 @@ impl EventCacheInner {
 
         // Left rooms.
         for (room_id, left_room_update) in updates.left {
-            let Ok(caches) = self.all_caches_for_room(&room_id).await else {
-                error!(?room_id, "Room must exist");
-                continue;
+            let caches = match self.all_caches_for_room(&room_id).await {
+                Ok(caches) => caches,
+                Err(err) => {
+                    self.drop_room_update(&room_id, err);
+                    continue;
+                }
             };
 
             if let Err(err) = caches.handle_left_room_update(left_room_update).await {
@@ -667,9 +703,12 @@ impl EventCacheInner {
         for (room_id, joined_room_update) in updates.joined {
             trace!(?room_id, "Handling a `JoinedRoomUpdate`");
 
-            let Ok(caches) = self.all_caches_for_room(&room_id).await else {
-                error!(?room_id, "Room must exist");
-                continue;
+            let caches = match self.all_caches_for_room(&room_id).await {
+                Ok(caches) => caches,
+                Err(err) => {
+                    self.drop_room_update(&room_id, err);
+                    continue;
+                }
             };
 
             if let Err(err) = caches.handle_joined_room_update(joined_room_update).await {
@@ -691,52 +730,105 @@ impl EventCacheInner {
     ) -> Result<OwnedRwLockReadGuard<CachesByRoom, Caches>> {
         // Fast path: the entry exists; let's acquire a read lock, it's cheaper than a
         // write lock.
-        match OwnedRwLockReadGuard::try_map(self.by_room.clone().read_owned().await, |by_room| {
+        if let Some(caches) = self.loaded_caches(room_id).await {
+            return Ok(caches);
+        }
+
+        // Slow-path: one creation per room at a time, run outside the room caches lock, so a
+        // creation that never finishes holds up its own room only.
+        let gate = self
+            .creation_gates
+            .lock()
+            .unwrap_or_else(PoisonError::into_inner)
+            .entry(room_id.to_owned())
+            .or_default()
+            .clone();
+        debug!(%room_id, "waiting to create a room's caches");
+        let _creating = gate.lock().await;
+
+        // In the meanwhile, some other caller might have done the same, so check for
+        // existence again.
+        if let Some(caches) = self.loaded_caches(room_id).await {
+            return Ok(caches);
+        }
+
+        let created = loop {
+            let clears = self.clears.load(Ordering::SeqCst);
+            let caches = match self.create_caches(room_id).await {
+                Ok(caches) => caches,
+                Err(err) => {
+                    self.creation_gates
+                        .lock()
+                        .unwrap_or_else(PoisonError::into_inner)
+                        .remove(room_id);
+                    return Err(err);
+                }
+            };
+            let mut by_room = self.by_room.clone().write_owned().await;
+            // a clear of every room that ran meanwhile wiped the store these caches were
+            // loaded from
+            if self.clears.load(Ordering::SeqCst) != clears {
+                continue;
+            }
+            by_room.insert(room_id.to_owned(), caches);
+            break by_room;
+        };
+        self.creation_gates.lock().unwrap_or_else(PoisonError::into_inner).remove(room_id);
+        debug!(%room_id, "created a room's caches");
+
+        Ok(OwnedRwLockWriteGuard::try_downgrade_map(created, |by_room| by_room.get(room_id))
+            .expect("`Caches` has just been inserted"))
+    }
+
+    fn drop_room_update(&self, room_id: &RoomId, err: EventCacheError) {
+        if matches!(err, EventCacheError::RoomCachesNotCreated { .. }) {
+            // the update is gone, so no later `pos` may claim it was stored
+            self.dropped_room_updates.store(true, Ordering::SeqCst);
+        }
+        error!(%room_id, "dropping a room update, its caches are unavailable: {err}");
+    }
+
+    async fn loaded_caches(
+        &self,
+        room_id: &RoomId,
+    ) -> Option<OwnedRwLockReadGuard<CachesByRoom, Caches>> {
+        OwnedRwLockReadGuard::try_map(self.by_room.clone().read_owned().await, |by_room| {
             by_room.get(room_id)
-        }) {
-            Ok(caches) => Ok(caches),
-
-            Err(by_room_guard) => {
-                // Slow-path: the entry doesn't exist; let's acquire a write lock.
-                drop(by_room_guard);
-                debug!(%room_id, "waiting for the room caches lock to create a room's caches");
-                let by_room_guard = self.by_room.clone().write_owned().await;
-
-                // In the meanwhile, some other caller might have obtained write access and done
-                // the same, so check for existence again.
-                let mut by_room_guard =
-                    match OwnedRwLockWriteGuard::try_downgrade_map(by_room_guard, |by_room| {
-                        by_room.get(room_id)
-                    }) {
-                        Ok(caches) => return Ok(caches),
-                        Err(by_room_guard) => by_room_guard,
-                    };
-
-                debug!(%room_id, "creating a room's caches");
-                let caches = Caches::new(
-                    &self.client,
-                    room_id,
-                    self.generic_update_sender.clone(),
-                    self.linked_chunk_update_sender.clone(),
-                    // SAFETY: we must have subscribed before reaching this code, otherwise
-                    // something is very wrong.
-                    self.auto_shrink_sender.get().cloned().expect(
-                        "we must have called `EventCache::subscribe()` before calling here.",
-                    ),
-                    self.store.clone(),
-                    self.automatic_pagination.get().cloned(),
-                )
-                .await?;
-                debug!(%room_id, "created a room's caches");
-
-                by_room_guard.insert(room_id.to_owned(), caches);
-
-                Ok(OwnedRwLockWriteGuard::try_downgrade_map(by_room_guard, |by_room| {
-                    by_room.get(room_id)
-                })
-                .expect("`Caches` has just been inserted"))
+        })
+        .ok()
+    }
+
+    /// Create a room's caches, abandoning an attempt that runs past [`CREATION_WAIT`] and
+    /// trying once more.
+    async fn create_caches(&self, room_id: &RoomId) -> Result<Caches> {
+        let mut stuck = "";
+        for attempt in 1..=2 {
+            debug!(%room_id, attempt, "creating a room's caches");
+            let step = caches::CreationStep::default();
+            let creation = Caches::new(
+                &self.client,
+                room_id,
+                self.generic_update_sender.clone(),
+                self.linked_chunk_update_sender.clone(),
+                // SAFETY: we must have subscribed before reaching this code, otherwise
+                // something is very wrong.
+                self.auto_shrink_sender
+                    .get()
+                    .cloned()
+                    .expect("we must have called `EventCache::subscribe()` before calling here."),
+                self.store.clone(),
+                self.automatic_pagination.get().cloned(),
+                &step,
+            );
+            match timeout(creation, CREATION_WAIT).await {
+                Ok(created) => return created,
+                Err(_) => {
+                    stuck = step.current();
+                    warn!(%room_id, attempt, step = stuck, "a room's caches weren't created in time");
+                }
             }
         }
+        Err(EventCacheError::RoomCachesNotCreated { room_id: room_id.to_owned(), step: stuck })
     }
 }

matrix-rust-sdk

Commit e4a8dbca1ac5

Commit `e4a8dbca1ac5`