mirror/matrix-rust-sdk
1
0
Fork 0
mirror of https://github.com/matrix-org/matrix-rust-sdk.git synced 2026-08-03 03:26:57 -04:00
Code Issues Packages Projects Releases Wiki Activity

refactor(sdk): Re-implement EventCacheInner::clear_all_rooms.

Browse Source 
This patch replaces `Caches::prepare_to_reset` and `ResetCaches` by a
new `EventCacheInner::clear_all_rooms` implementation. Now, it folds to
two steps:

1. Take an exclusive lock guard over all the caches of all the rooms,
2. Clear and reload all states with the new `State` structure.

It greatly simplifies this feature.
This commit is contained in:
Ivan Enderlin
2026-06-05 13:42:14 +02:00
parent 0c83e653c0
commit fe24a2590f
4 changed files with 62 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
crates/matrix-sdk-base/src/event_cache/store/mod.rs
View File

@@ -125,7 +125,10 @@ impl EventCacheStoreLockGuard {
this.cross_process_lock_guard.clear_dirty();
}
/// Force to [`CrossProcessLockGuard::is_dirty`].
/// Forward to [`CrossProcessLockGuard::is_dirty`].
///
/// This is an associated method to avoid colliding with the [`Deref`]
/// implementation.
pub fn is_dirty(this: &Self) -> bool {
this.cross_process_lock_guard.is_dirty()
}

111
crates/matrix-sdk/src/event_cache/caches/mod.rs
View File

@@ -408,17 +408,6 @@ impl Caches {
Ok(())
}
/// Try to acquire exclusive locks over all the event caches managed by
/// this [`Caches`], in order to reset all the in-memory data.
///
/// Note that this method takes `&mut self`, ensuring only one reset can
/// happen at a time.
///
/// If the returned value is dropped, no data will be reset.
pub async fn prepare_to_reset(&mut self) -> Result<ResetCaches<'_>> {
ResetCaches::new(self).await
}
/// Get all events from all the event caches manged by this [`Caches`].
///
/// Events can be duplicated if present in different event caches.
@@ -430,106 +419,6 @@ impl Caches {
}
}
/// Type holding exclusive locks over all event caches managed by a
/// [`Caches`].
///
/// To reset all the event caches, call [`ResetCaches::reset_all`]. If this type
/// is dropped, no reset happens and the exclusive lock is released.
pub(super) struct ResetCaches<'c> {}
impl<'c> ResetCaches<'c> {
/// Create a new [`ResetCaches`].
///
/// It can fail if acquiring an exclusive lock fails.
async fn new(
Caches { room, threads, pinned_events, event_focused, internals: _ }: &'c mut Caches,
) -> Result<Self> {
todo!()
}
/// Reset all the event caches, and broadcast the [`TimelineVectorDiffs`].
///
/// Note that this method consumes `self`, ensuring the acquired exclusive
/// locks over the event caches are released.
///
/// It can fail if resetting an event cache fails.
pub async fn reset_all(self) -> Result<()> {
todo!()
/*
let Self {
room_lock,
threads_lock,
thread_locks,
pinned_events_lock,
event_focused_lock,
event_focused_locks,
} = self;
// Room.
{
let (mut room_state, room_update_sender) = room_lock;
let updates_as_vector_diffs = room_state.reset().await?;
room_update_sender.send(
room::RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
diffs: updates_as_vector_diffs,
origin: EventsOrigin::Cache,
}),
Some(room::RoomEventCacheGenericUpdate { room_id: room_state.room_id.clone() }),
);
}
// Threads.
{
for (mut thread_state, thread_update_sender) in thread_locks {
let updates_as_vector_diffs = thread_state.reset().await?;
thread_update_sender.send(
TimelineVectorDiffs {
diffs: updates_as_vector_diffs,
origin: EventsOrigin::Cache,
},
// This function is part of the `RoomEventCache` flow. The generic update is
// handled by it.
None,
);
}
// Now we can release the exclusive access over the threads.
drop(threads_lock);
}
// Pinned-events.
{
if let Some((mut pinned_events_state, pinned_events_update_sender)) = pinned_events_lock
{
let updates_as_vector_diffs = pinned_events_state.reset().await?;
pinned_events_update_sender.send(TimelineVectorDiffs {
diffs: updates_as_vector_diffs,
origin: EventsOrigin::Cache,
});
}
}
// Event-focused.
{
for (mut event_focused_state, event_focused_update_sender) in event_focused_locks {
let updates_as_vector_diffs = event_focused_state.reset()?;
let _ = event_focused_update_sender.send(TimelineVectorDiffs {
diffs: updates_as_vector_diffs,
origin: EventsOrigin::Cache,
});
}
// Now we can release the exclusive access over the event-focused caches.
drop(event_focused_lock);
}
Ok(())
*/
}
}
/// A diff update for an event cache timeline represented as a vector.
#[derive(Clone, Debug)]
pub struct TimelineVectorDiffs {

70
crates/matrix-sdk/src/event_cache/mod.rs
View File

@@ -639,63 +639,33 @@ impl EventCacheInner {
/// Clears all the room's data.
async fn clear_all_rooms(&self) -> Result<()> {
// Okay, here's where things get complicated.
// Okay, here's where things get delicate.
//
// On the one hand, `by_room` may include storage for *some* rooms that we know
// about, but not *all* of them. Any room that hasn't been loaded in the
// client, or touched by a sync, will remain unloaded in memory, so it
// will be missing from `self.by_room`. As a result, we need to make
// sure that we're hitting the storage backend to *really* clear all the
// rooms, including those that haven't been loaded yet.
// On the one hand, `by_room` may include storage for *some* caches
// that we know about, but not *all* of them. Any cache that hasn't been
// loaded in the client, or touched by a sync, will remain unloaded in
// memory, so it will be missing from `self.by_room`. As a result, we
// need to make sure that we're hitting the storage backend to *really*
// clear all the caches, including those that haven't been loaded yet.
//
// On the other hand, one must NOT clear the `by_room` map, because if someone
// subscribed to a room update, they would never get any new update for
// that room, since re-creating the `RoomEventCache` would create a new,
// unrelated sender.
// On the other hand, one must NOT clear the `by_room` map, because if
// someone subscribed to a cache update, they would never get any new
// update for that cache, since re-creating the cache would create a
// new, unrelated sender.
//
// So we need to *keep* the rooms in `by_room` alive, while clearing them in the
// store backend.
// So we need to *keep* the caches in `by_room` alive, while clearing
// them in the store backend.
//
// As a result, for a short while, the in-memory linked chunks
// will be desynchronized from the storage. We need to be careful then. During
// that short while, we don't want *anyone* to touch the linked chunk
// As a result, for a short while, the in-memory linked chunks will be
// desynchronised from the storage. We need to be careful then. During
// that short while, we don't want *anyone* to touch the linked chunks
// (be it in memory or in the storage).
//
// And since that requirement applies to *any* room in `by_room` at the same
// time, we'll have to take the locks for *all* the live rooms, so as to
// properly clear the underlying storage.
//
// At this point, you might be scared about the potential for deadlocking. I am
// as well, but I'm convinced we're fine:
//
// 1. the lock for `by_room` is usually held only for a short while, and
// independently of the other two kinds.
// 2. the state may acquire the store cross-process lock internally, but only
// while the state's methods are called (so it's always transient). As a
// result, as soon as we've acquired the state locks, the store lock ought to
// be free.
// 3. The store lock is held explicitly only in a small scoped area below.
// 4. Then the store lock will be held internally when calling `reset_all()`,
// but at this point it's only held for a short while each time, so rooms
// will take turn to acquire it.
// And since that requirement applies to *any* cache in `by_room` at the
// same time, we'll have to take the lock for *all* the live caches and
// for the states, so as to properly clear the underlying storage.
let mut all_caches = self.by_room.write().await;
// Prepare to reset all the caches: it ensures nobody is accessing or mutating
// them.
let resets =
try_join_all(all_caches.values_mut().map(|caches| caches.prepare_to_reset())).await?;
// Clear the storage for all the rooms, using the storage facility.
let store_guard = match self.store.lock().await? {
EventCacheStoreLockState::Clean(store_guard) => store_guard,
EventCacheStoreLockState::Dirty(store_guard) => store_guard,
};
store_guard.clear_all_linked_chunks().await?;
// At this point, all the in-memory linked chunks are desynchronized from their
// storages. Resynchronize them manually by resetting them.
try_join_all(resets.into_iter().map(|reset_cache| reset_cache.reset_all())).await?;
self.state.clear_and_reload(self.by_room.write().await).await?;
Ok(())
}

40
crates/matrix-sdk/src/event_cache/states/mod.rs
View File

@@ -28,7 +28,7 @@ use ruma::{OwnedEventId, OwnedRoomId};
use tokio::sync::{Mutex, RwLock, RwLockMappedWriteGuard, RwLockReadGuard, RwLockWriteGuard};
use super::{
EventCacheError, EventsOrigin, Result,
CachesByRoom, EventCacheError, EventsOrigin, Result,
caches::{
TimelineVectorDiffs,
event_focused::{EventFocusedCacheKey, EventFocusedCacheState},
@@ -184,7 +184,7 @@ impl StateLock {
/// dropped.
///
/// If the cross-process lock over the store is dirty (see
/// [`EventCacheStoreLockState`]), the state is reloaded.
/// [`EventCacheStoreLockState`]), the state is reloaded automatically.
async fn write<'state>(&'state self) -> Result<ReloadableStateLockWriteGuard<'state>> {
let state_guard = self.inner.locked_state.write().await;
@@ -207,6 +207,42 @@ impl StateLock {
})
}
/// Clear and reload all states: in-memory and in-store.
///
/// The `caches_for_all_rooms_exclusive_lock_guard` argument ensures an
/// exclusive lock over all the caches has been acquired. This is required
/// to ensure safety for this method.
pub(super) async fn clear_and_reload(
&self,
caches_for_all_rooms_exclusive_lock_guard: RwLockWriteGuard<'_, CachesByRoom>,
) -> Result<()> {
let state_guard = self.inner.locked_state.write().await;
let mut guard = match state_guard.store.lock().await? {
EventCacheStoreLockState::Clean(store_guard)
| EventCacheStoreLockState::Dirty(store_guard) => {
ReloadableStateLockWriteGuard { state: state_guard, store: store_guard }
}
};
// Clear the storage for all the caches for all rooms.
guard.store.clear_all_linked_chunks().await?;
// At this point, all the in-memory linked chunks are desynchronised
// from the storage. Resynchronise them manually by reloading them.
guard.reload().await?;
if EventCacheStoreLockGuard::is_dirty(&guard.store) {
// All good because the state has been reloaded, mark the
// cross-process lock as non-dirty.
EventCacheStoreLockGuard::clear_dirty(&guard.store);
}
drop(caches_for_all_rooms_exclusive_lock_guard);
Ok(())
}
/// Insert a new cache state at location `cache_state_selector` if none
/// exists.
///