matrix-rust-sdk

public

Star0

Fork 0

Watch0

56 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `29016b0c0bae`

tarball

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

2 files changed+14-20

event-cache: hand out room caches as clones, never as a guard over the room caches map

e4a8dbca made creation take the room caches write lock only to insert, but `all_caches_for_room` still returned a read guard, and sync held one for a room's whole update, store writes included. In 16 of 60 soak runs a creation that finished meanwhile never logged its insert and sync never reached its next room, which is what a writer stuck behind a held read guard looks like under tokio's fair lock, with every later reader queued behind the writer. `Caches` is a shallow clone, so callers now get one and the map is held only while it is read or written.

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

@@ -117,7 +117,7 @@ impl CreationStep {

117117 }

118118

119119 /// A type to hold all the caches for a given room.

120-#[derive(Debug)]

120+#[derive(Clone, Debug)]

121121 pub(super) struct Caches {

122122 pub room: room::RoomEventCache,

123123 }

Modified file: crates/matrix-sdk/src/event_cache/mod.rs+13 additions−19 deletions@@ -50,7 +50,7 @@ use matrix_sdk_base::{
 use matrix_sdk_common::timeout::timeout;
 use ruma::{OwnedRoomId, RoomId};
 use tokio::sync::{
-    Mutex, OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock,
+    Mutex, RwLock,
     broadcast::{Receiver, Sender, channel},
     mpsc, watch,
 };
@@ -416,7 +416,7 @@ impl EventCache {
             return Err(EventCacheError::NotSubscribedYet);
         };
 
-        let room = self.inner.all_caches_for_room(room_id).await?.room.clone();
+        let room = self.inner.all_caches_for_room(room_id).await?.room;
 
         Ok((room, drop_handles))
     }
@@ -724,10 +724,11 @@ impl EventCacheInner {
     }
 
     /// Return all the event caches associated to a specific room.
-    async fn all_caches_for_room(
-        &self,
-        room_id: &RoomId,
-    ) -> Result<OwnedRwLockReadGuard<CachesByRoom, Caches>> {
+    ///
+    /// A shallow clone rather than a guard. Sync handles a room for as long as its
+    /// store writes take, and a guard held that long left a finished creation waiting
+    /// to insert, with every later reader queued behind it.
+    async fn all_caches_for_room(&self, room_id: &RoomId) -> Result<Caches> {
         // Fast path: the entry exists; let's acquire a read lock, it's cheaper than a
         // write lock.
         if let Some(caches) = self.loaded_caches(room_id).await {
@@ -764,20 +765,19 @@ impl EventCacheInner {
                     return Err(err);
                 }
             };
-            let mut by_room = self.by_room.clone().write_owned().await;
+            let mut by_room = self.by_room.write().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;
+            by_room.insert(room_id.to_owned(), caches.clone());
+            break caches;
         };
         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"))
+        Ok(created)
     }
 
     fn drop_room_update(&self, room_id: &RoomId, err: EventCacheError) {
@@ -788,14 +788,8 @@ impl EventCacheInner {
         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()
+    async fn loaded_caches(&self, room_id: &RoomId) -> Option<Caches> {
+        self.by_room.read().await.get(room_id).cloned()
     }
 
     /// Create a room's caches, abandoning an attempt that runs past [`CREATION_WAIT`] and

matrix-rust-sdk

Commit 29016b0c0bae

Commit `29016b0c0bae`