matrix-rust-sdk

public

Star0

Fork 0

Watch0

50 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `ae6eba54f24b`

tarball

atagen <boss@atagen.co>2026-10-27 05:59 UTC

1 file changed+139-17

event-cache: name who holds a cache state lock when a wait on it runs long

A room open on wasm can wedge with the history restore never reaching its first IndexedDB call while no IndexedDB transaction is open, so it waits on the room state lock behind a holder suspended elsewhere, and nothing in a trace says which. Each guard now records its caller, and a wait past 5s warns with every current holder and how long each has held the lock.

Modified file: crates/matrix-sdk/src/event_cache/caches/lock.rs+139 additions−17 deletions@@ -19,12 +19,21 @@
 //!
 //! [`RoomEventCache`]: super::super::RoomEventCache
 
-use std::ops::{Deref, DerefMut};
+use std::{
+    future::Future,
+    ops::{Deref, DerefMut},
+    panic::Location,
+    sync::Mutex as StdMutex,
+    time::Duration,
+};
 
 use matrix_sdk_base::event_cache::store::{
     EventCacheStoreLock, EventCacheStoreLockGuard, EventCacheStoreLockState,
 };
+use matrix_sdk_common::timeout::timeout;
+use ruma::time::Instant;
 use tokio::sync::{Mutex, RwLock, RwLockWriteGuard};
+use tracing::warn;
 
 use super::super::Result;
 
@@ -42,12 +51,95 @@ pub struct StateLock<S> {
     /// Please see inline comment of [`Self::read`] to understand why it
     /// exists.
     state_lock_upgrade_mutex: Mutex<()>,
+
+    holders: StdMutex<Holders>,
+}
+
+/// How long a wait on a state lock runs before it names whoever holds the lock.
+const SLOW_WAIT: Duration = Duration::from_secs(5);
+
+#[derive(Default)]
+struct Holders {
+    next: u64,
+    held: Vec<(u64, &'static Location<'static>, &'static str, Instant)>,
+}
+
+struct Holder<'a> {
+    id: u64,
+    holders: &'a StdMutex<Holders>,
+}
+
+impl Drop for Holder<'_> {
+    fn drop(&mut self) {
+        let mut holders = self.holders.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        holders.held.retain(|(id, ..)| *id != self.id);
+    }
 }
 
 impl<S> StateLock<S> {
     /// Construct a new [`StateLock`].
     pub fn new_inner(state: S) -> Self {
-        Self { locked_state: RwLock::new(state), state_lock_upgrade_mutex: Mutex::new(()) }
+        Self {
+            locked_state: RwLock::new(state),
+            state_lock_upgrade_mutex: Mutex::new(()),
+            holders: Default::default(),
+        }
+    }
+
+    fn hold(&self, caller: &'static Location<'static>, kind: &'static str) -> Holder<'_> {
+        let mut holders = self.holders.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+        holders.next += 1;
+        let id = holders.next;
+        holders.held.push((id, caller, kind, Instant::now()));
+        Holder { id, holders: &self.holders }
+    }
+
+    async fn waiting<T>(
+        &self,
+        caller: &'static Location<'static>,
+        kind: &'static str,
+        acquire: impl Future<Output = T>,
+    ) -> T {
+        let mut acquire = std::pin::pin!(acquire);
+        match timeout(acquire.as_mut(), SLOW_WAIT).await {
+            Ok(acquired) => acquired,
+            Err(_) => {
+                let held = {
+                    let holders =
+                        self.holders.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
+                    holders
+                        .held
+                        .iter()
+                        .map(|(_, at, kind, since)| {
+                            format!("{kind} at {at} for {:?}", since.elapsed())
+                        })
+                        .collect::<Vec<_>>()
+                        .join(", ")
+                };
+                warn!(%caller, kind, %held, "still waiting on a cache state lock");
+                acquire.await
+            }
+        }
+    }
+
+    /// Lock this [`StateLock`] with per-thread shared access.
+    #[track_caller]
+    pub fn read<'a>(&'a self) -> impl Future<Output = Result<StateLockReadGuard<'a, S>>> + 'a
+    where
+        S: Store,
+        StateLockWriteGuard<'a, S>: Reload,
+    {
+        self.read_at(Location::caller())
+    }
+
+    /// Lock this [`StateLock`] with exclusive per-thread write access.
+    #[track_caller]
+    pub fn write<'a>(&'a self) -> impl Future<Output = Result<StateLockWriteGuard<'a, S>>> + 'a
+    where
+        S: Store,
+        StateLockWriteGuard<'a, S>: Reload,
+    {
+        self.write_at(Location::caller())
     }
 
     /// Lock this [`StateLock`] with per-thread shared access.
@@ -59,7 +151,10 @@ impl<S> StateLock<S> {
     ///
     /// If the cross-process lock over the store is dirty (see
     /// [`EventCacheStoreLockState`]), the state is reset to the last chunk.
-    pub async fn read<'a>(&'a self) -> Result<StateLockReadGuard<'a, S>>
+    async fn read_at<'a>(
+        &'a self,
+        caller: &'static Location<'static>,
+    ) -> Result<StateLockReadGuard<'a, S>>
     where
         S: Store,
         StateLockWriteGuard<'a, S>: Reload,
@@ -110,15 +205,21 @@ impl<S> StateLock<S> {
         //
         // [^1]: https://docs.rs/lock_api/0.4.14/lock_api/struct.RwLock.html#method.upgradable_read
         // [^2]: https://docs.rs/async-lock/3.4.1/async_lock/struct.RwLock.html#method.upgradable_read
-        let _state_lock_upgrade_guard = self.state_lock_upgrade_mutex.lock().await;
+        let (_state_lock_upgrade_guard, state_guard) = self
+            .waiting(caller, "read", async {
+                let upgrade = self.state_lock_upgrade_mutex.lock().await;
 
-        // Obtain a read lock.
-        let state_guard = self.locked_state.read().await;
+                // Obtain a read lock.
+                (upgrade, self.locked_state.read().await)
+            })
+            .await;
 
         match state_guard.store().lock().await? {
-            EventCacheStoreLockState::Clean(store_guard) => {
-                Ok(StateLockReadGuard { state: state_guard, store: store_guard })
-            }
+            EventCacheStoreLockState::Clean(store_guard) => Ok(StateLockReadGuard {
+                state: state_guard,
+                store: store_guard,
+                _holder: self.hold(caller, "read"),
+            }),
             EventCacheStoreLockState::Dirty(store_guard) => {
                 // Drop the read lock, and take a write lock to modify the state.
                 // This is safe because only one reader at a time (see
@@ -126,7 +227,11 @@ impl<S> StateLock<S> {
                 drop(state_guard);
                 let state_guard = self.locked_state.write().await;
 
-                let mut guard = StateLockWriteGuard { state: state_guard, store: store_guard };
+                let mut guard = StateLockWriteGuard {
+                    state: state_guard,
+                    store: store_guard,
+                    _holder: self.hold(caller, "read"),
+                };
 
                 // Reload the state.
                 guard.reload().await?;
@@ -151,19 +256,28 @@ impl<S> StateLock<S> {
     ///
     /// If the cross-process lock over the store is dirty (see
     /// [`EventCacheStoreLockState`]), the state is reset to the last chunk.
-    pub async fn write<'a>(&'a self) -> Result<StateLockWriteGuard<'a, S>>
+    async fn write_at<'a>(
+        &'a self,
+        caller: &'static Location<'static>,
+    ) -> Result<StateLockWriteGuard<'a, S>>
     where
         S: Store,
         StateLockWriteGuard<'a, S>: Reload,
     {
-        let state_guard = self.locked_state.write().await;
+        let state_guard = self.waiting(caller, "write", self.locked_state.write()).await;
 
         match state_guard.store().lock().await? {
-            EventCacheStoreLockState::Clean(store_guard) => {
-                Ok(StateLockWriteGuard { state: state_guard, store: store_guard })
-            }
+            EventCacheStoreLockState::Clean(store_guard) => Ok(StateLockWriteGuard {
+                state: state_guard,
+                store: store_guard,
+                _holder: self.hold(caller, "write"),
+            }),
             EventCacheStoreLockState::Dirty(store_guard) => {
-                let mut guard = StateLockWriteGuard { state: state_guard, store: store_guard };
+                let mut guard = StateLockWriteGuard {
+                    state: state_guard,
+                    store: store_guard,
+                    _holder: self.hold(caller, "write"),
+                };
 
                 // Reload the state.
                 guard.reload().await?;
@@ -184,6 +298,8 @@ pub struct StateLockReadGuard<'a, S> {
 
     /// The cross-process lock guard over the store.
     pub store: EventCacheStoreLockGuard,
+
+    _holder: Holder<'a>,
 }
 
 impl<'a, S> Deref for StateLockReadGuard<'a, S> {
@@ -201,6 +317,8 @@ pub struct StateLockWriteGuard<'a, S> {
 
     /// The cross-process lock guard over the store.
     pub store: EventCacheStoreLockGuard,
+
+    _holder: Holder<'a>,
 }
 
 impl<'a, S> Deref for StateLockWriteGuard<'a, S> {
@@ -226,7 +344,11 @@ impl<'a, S> StateLockWriteGuard<'a, S> {
     /// It returns an RAII guard which will drop the read access to the
     /// state and to the store when dropped.
     fn downgrade(self) -> StateLockReadGuard<'a, S> {
-        StateLockReadGuard { state: self.state.downgrade(), store: self.store }
+        StateLockReadGuard {
+            state: self.state.downgrade(),
+            store: self.store,
+            _holder: self._holder,
+        }
     }
 }

matrix-rust-sdk

Commit ae6eba54f24b

Commit `ae6eba54f24b`