matrix-rust-sdk

public

Star0

Fork 0

Watch0

51 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `82e22202f3da`

tarball

atagen <boss@atagen.co>2026-10-27 06:25 UTC

4 files changed+59-42

event-cache: stop a stalled event save from holding the room state

`RoomEventCache::save_events` held the room state write lock while it waited on a spawned `save_event`. On wasm that save can stall with no IndexedDB transaction open, and the lock holder report caught one at room open, behind a `Room` fetch that saves what it got, wedging the history restore and every later sync of that room behind it. The save only touches the store, whose single transaction keeps each read and write together, so it now runs after the lock is released. Every save also stops waiting after 10s, so the saves that still run under the lock during sync post-processing are bounded too.

Modified file: crates/matrix-sdk/src/event_cache/caches/mod.rs+42 additions−2 deletions@@ -12,18 +12,26 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use std::sync::{Arc, atomic::AtomicU64};
+use std::{
+    sync::{Arc, atomic::AtomicU64},
+    time::Duration,
+};
 
 use eyeball::SharedObservable;
 use eyeball_im::VectorDiff;
 use matrix_sdk_base::{
     ThreadingSupport,
-    event_cache::{Event, store::EventCacheStoreLock},
+    event_cache::{
+        Event,
+        store::{EventCacheStoreLock, EventCacheStoreLockGuard},
+    },
     linked_chunk::Position,
     sync::{JoinedRoomUpdate, LeftRoomUpdate},
 };
+use matrix_sdk_common::{executor::spawn, timeout::timeout};
 use ruma::{OwnedRoomId, RoomId};
 use tokio::sync::{broadcast::Sender, mpsc};
+use tracing::{debug, warn};
 
 use super::{EventCacheError, EventsOrigin, Result};
 use crate::{
@@ -39,6 +47,38 @@ mod read_receipts;
 pub mod room;
 pub mod thread;
 
+/// How long a caller waits for events to reach the store.
+const SAVE_WAIT: Duration = Duration::from_secs(10);
+
+/// Save events into the store, without notifying observers.
+///
+/// The save runs in its own task so cancelling the caller can't cut it short, and the caller
+/// stops waiting after [`SAVE_WAIT`] so a save that never finishes can't hold a cache lock.
+pub(super) async fn save_events(
+    store: EventCacheStoreLockGuard,
+    room_id: OwnedRoomId,
+    events: Vec<Event>,
+) -> Result<()> {
+    let count = events.len();
+    let saving = spawn({
+        let room_id = room_id.clone();
+        async move {
+            debug!(%room_id, count, "saving events to the store");
+            for event in events {
+                store.save_event(&room_id, event).await?;
+            }
+            Result::Ok(())
+        }
+    });
+    match timeout(saving, SAVE_WAIT).await {
+        Ok(joined) => joined.expect("joining failed"),
+        Err(_) => {
+            warn!(%room_id, count, "events not saved to the store yet, no longer waiting");
+            Ok(())
+        }
+    }
+}
+
 /// A type to hold all the caches for a given room.
 #[derive(Debug)]
 pub(super) struct Caches {

Modified file: crates/matrix-sdk/src/event_cache/caches/room/mod.rs+9 additions−7 deletions@@ -429,16 +429,18 @@ impl RoomEventCache {
     /// Save some events in the event cache, for further retrieval with
     /// [`Self::event`].
     pub(crate) async fn save_events(&self, events: impl IntoIterator<Item = Event>) {
-        match self.inner.state.write().await {
-            Ok(mut state_guard) => {
-                if let Err(err) = state_guard.save_events(events).await {
-                    warn!("couldn't save event in the event cache: {err}");
-                }
-            }
-
+        // the save touches only the store, which keeps each save's read and write in one
+        // transaction, so holding the room's state across it only wedged sync behind it
+        let store = match self.inner.state.read().await {
+            Ok(state_guard) => state_guard.store_guard(),
             Err(err) => {
                 warn!("couldn't save event in the event cache: {err}");
+                return;
             }
+        };
+        let events = events.into_iter().collect();
+        if let Err(err) = super::save_events(store, self.inner.room_id.clone(), events).await {
+            warn!("couldn't save event in the event cache: {err}");
         }
     }
 

Modified file: crates/matrix-sdk/src/event_cache/caches/room/state.rs+6 additions−16 deletions@@ -36,7 +36,6 @@ use matrix_sdk_base::{
     serde_helpers::{extract_edit_target, extract_thread_root},
     sync::Timeline,
 };
-use matrix_sdk_common::executor::spawn;
 use ruma::{
     EventId, OwnedEventId, OwnedRoomId, OwnedUserId, TransactionId,
     events::{
@@ -462,6 +461,10 @@ impl<'a> RoomEventCacheStateLockReadGuard<'a> {
         super::history::gap_positions(&self.state.room_linked_chunk)
     }
 
+    pub(super) fn store_guard(&self) -> EventCacheStoreLockGuard {
+        self.store.clone()
+    }
+
     pub(super) fn has_sync_hole(&self) -> bool {
         self.state.sync_hole.is_some()
     }
@@ -1624,21 +1627,8 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
         &mut self,
         events: impl IntoIterator<Item = Event>,
     ) -> Result<(), EventCacheError> {
-        let store = self.store.clone();
-        let room_id = self.state.room_id.clone();
-        let events = events.into_iter().collect::<Vec<_>>();
-
-        // Spawn a task so the save is uninterrupted by task cancellation.
-        spawn(async move {
-            for event in events {
-                store.save_event(&room_id, event).await?;
-            }
-            super::Result::Ok(())
-        })
-        .await
-        .expect("joining failed")?;
-
-        Ok(())
+        let events = events.into_iter().collect();
+        super::super::save_events(self.store.clone(), self.state.room_id.clone(), events).await
     }
 
     #[cfg(test)]

Modified file: crates/matrix-sdk/src/event_cache/caches/thread/state.rs+2 additions−17 deletions@@ -19,7 +19,6 @@ use matrix_sdk_base::{
     event_cache::{Event, Gap, store::EventCacheStoreLock},
     linked_chunk::{OwnedLinkedChunkId, Position, Update},
 };
-use matrix_sdk_common::executor::spawn;
 use ruma::{OwnedEventId, OwnedRoomId};
 use tokio::sync::broadcast::Sender;
 use tracing::instrument;
@@ -192,22 +191,8 @@ impl<'a> ThreadEventCacheStateLockWriteGuard<'a> {
 
     /// Save events into the database, without notifying observers.
     pub async fn save_events(&mut self, events: impl IntoIterator<Item = Event>) -> Result<()> {
-        let store = self.store.clone();
-        let room_id = self.state.room_id.clone();
-        let events = events.into_iter().collect::<Vec<_>>();
-
-        // Spawn a task so the save is uninterrupted by task cancellation.
-        spawn(async move {
-            for event in events {
-                store.save_event(&room_id, event).await?;
-            }
-
-            Result::Ok(())
-        })
-        .await
-        .expect("joining failed")?;
-
-        Ok(())
+        let events = events.into_iter().collect();
+        super::super::save_events(self.store.clone(), self.state.room_id.clone(), events).await
     }
 
     /// Reset this data structure as if it were brand new.

matrix-rust-sdk

Commit 82e22202f3da

Commit `82e22202f3da`