diff --git a/crates/matrix-sdk/src/event_cache/caches/aggregator.rs b/crates/matrix-sdk/src/event_cache/caches/aggregator.rs index ba659e9a9..f5c454015 100644 --- a/crates/matrix-sdk/src/event_cache/caches/aggregator.rs +++ b/crates/matrix-sdk/src/event_cache/caches/aggregator.rs @@ -20,7 +20,11 @@ use matrix_sdk_base::{ }; use ruma::{OwnedEventId, events::relation::RelationType, room_version_rules::RedactionRules}; -use super::{super::Result, room::RoomEventCacheStateLockReadGuard, thread::ThreadEventCache}; +use super::{ + super::{Result, states::StateLockReadGuard}, + room::RoomEventCacheState, + thread::ThreadEventCache, +}; pub fn aggregate_timeline_for_room(timeline: Timeline) -> Timeline { timeline @@ -29,7 +33,7 @@ pub fn aggregate_timeline_for_room(timeline: Timeline) -> Timeline { pub async fn aggregate_timeline_for_threads( timeline: &Timeline, existing_threads: &HashMap, - room_event_cache: RoomEventCacheStateLockReadGuard<'_>, + room_event_cache: StateLockReadGuard<'_, RoomEventCacheState>, redaction_rules: &RedactionRules, ) -> Result> { let mut new_events_by_thread = HashMap::new(); diff --git a/crates/matrix-sdk/src/event_cache/caches/mod.rs b/crates/matrix-sdk/src/event_cache/caches/mod.rs index da59a2e98..2ed15d62c 100644 --- a/crates/matrix-sdk/src/event_cache/caches/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/mod.rs @@ -18,7 +18,7 @@ use eyeball::SharedObservable; use eyeball_im::VectorDiff; use matrix_sdk_base::{ ThreadingSupport, - event_cache::{Event, store::EventCacheStoreLock}, + event_cache::Event, linked_chunk::Position, sync::{JoinedRoomUpdate, LeftRoomUpdate}, }; @@ -26,7 +26,9 @@ use once_cell::sync::OnceCell; use ruma::{OwnedEventId, OwnedRoomId, RoomId, room_version_rules::RoomVersionRules}; use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock, broadcast::Sender, mpsc}; -use super::{EventCacheError, EventsOrigin, Result, automatic_pagination::AutomaticPagination}; +use super::{ + EventCacheError, EventsOrigin, Result, automatic_pagination::AutomaticPagination, states, +}; use crate::{client::WeakClient, room::WeakRoom}; mod aggregator; @@ -71,7 +73,7 @@ pub(super) struct Caches { #[derive(Debug)] struct CachesInternals { - store: EventCacheStoreLock, + state: states::StateLock, linked_chunk_update_sender: Sender, room_version_rules: RoomVersionRules, } @@ -84,7 +86,7 @@ impl Caches { generic_update_sender: Sender, linked_chunk_update_sender: Sender, auto_shrink_sender: mpsc::Sender, - store: EventCacheStoreLock, + state: &states::StateLock, automatic_pagination: Option, ) -> Result { let Some(client) = weak_client.get() else { @@ -110,19 +112,25 @@ impl Caches { let own_user_id = client.user_id().expect("the user must be logged in, at this point").to_owned(); - let room_state = room::LockedRoomEventCacheState::new( - own_user_id.clone(), - room_id.to_owned(), - weak_room.clone(), - room_version_rules.clone(), - enabled_thread_support, - update_sender.clone(), - linked_chunk_update_sender.clone(), - store.clone(), - pagination_status.clone(), - automatic_pagination, - ) - .await?; + let room_state = state + .try_insert_once_with( + states::selectors::RoomStateSelector::new(room_id.to_owned()), + |store_guard| { + room::RoomEventCacheState::new( + own_user_id.clone(), + room_id.to_owned(), + weak_room.clone(), + room_version_rules.clone(), + enabled_thread_support, + update_sender.clone(), + linked_chunk_update_sender.clone(), + store_guard, + pagination_status.clone(), + automatic_pagination, + ) + }, + ) + .await?; let timeline_is_not_empty = room_state.read().await?.room_linked_chunk().revents().next().is_some(); @@ -149,7 +157,11 @@ impl Caches { threads: Arc::new(RwLock::new(HashMap::new())), pinned_events: OnceCell::new(), event_focused: Arc::new(RwLock::new(HashMap::new())), - internals: CachesInternals { store, linked_chunk_update_sender, room_version_rules }, + internals: CachesInternals { + state: state.clone(), + linked_chunk_update_sender, + room_version_rules, + }, }) } @@ -192,7 +204,8 @@ impl Caches { room.own_user_id().to_owned(), self.internals.room_version_rules.clone(), room.weak_room().to_owned(), - self.internals.store.clone(), + // self.internals.store.clone(), + todo!(), room.update_sender().generic_update_sender().clone(), self.internals.linked_chunk_update_sender.clone(), ) @@ -218,7 +231,8 @@ impl Caches { self.room.own_user_id().clone(), self.internals.room_version_rules.clone(), self.internals.linked_chunk_update_sender.clone(), - self.internals.store.clone(), + // self.internals.store.clone(), + todo!(), ) }) } @@ -420,7 +434,10 @@ impl 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> { - room_lock: (room::RoomEventCacheStateLockWriteGuard<'c>, room::RoomEventCacheUpdateSender), + room_lock: ( + states::StateLockWriteGuard<'c, room::RoomEventCacheState>, + room::RoomEventCacheUpdateSender, + ), threads_lock: OwnedRwLockWriteGuard>, thread_locks: Vec<( thread::OwnedThreadEventCacheStateLockWriteGuard, diff --git a/crates/matrix-sdk/src/event_cache/caches/room/mod.rs b/crates/matrix-sdk/src/event_cache/caches/room/mod.rs index e8868b39a..083776420 100644 --- a/crates/matrix-sdk/src/event_cache/caches/room/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/room/mod.rs @@ -38,10 +38,8 @@ use tokio::sync::{Notify, mpsc}; use tracing::{instrument, trace, warn}; use self::pagination::RoomPagination; -pub(super) use self::state::{ - LockedRoomEventCacheState, RoomEventCacheStateLockReadGuard, RoomEventCacheStateLockWriteGuard, -}; pub use self::{ + state::RoomEventCacheState, subscriber::RoomEventCacheSubscriber, updates::{ RoomEventCacheGenericUpdate, RoomEventCacheLinkedChunkUpdate, RoomEventCacheUpdate, @@ -49,7 +47,10 @@ pub use self::{ }, }; use super::{ - super::{AutoShrinkChannelPayload, EventsOrigin, Result}, + super::{ + AutoShrinkChannelPayload, EventsOrigin, Result, + states::{CacheStateLock, StateLockWriteGuard, selectors::RoomStateSelector}, + }, TimelineVectorDiffs, event_linked_chunk::sort_positions_descending, pagination::SharedPaginationStatus, @@ -76,7 +77,7 @@ impl RoomEventCache { room_id: OwnedRoomId, weak_room: WeakRoom, own_user_id: OwnedUserId, - state: LockedRoomEventCacheState, + state: CacheStateLock, shared_pagination_status: SharedObservable, auto_shrink_sender: mpsc::Sender, update_sender: RoomEventCacheUpdateSender, @@ -231,7 +232,7 @@ impl RoomEventCache { } /// Return a reference to the state. - pub(in super::super) fn state(&self) -> &LockedRoomEventCacheState { + pub(in super::super) fn state(&self) -> &CacheStateLock { &self.inner.state } @@ -337,8 +338,8 @@ pub(super) struct RoomEventCacheInner { /// The user's own user id. own_user_id: OwnedUserId, - /// State for this room's event cache. - state: LockedRoomEventCacheState, + /// State for this room's cache. + state: CacheStateLock, /// A notifier that we received a new pagination token. pagination_batch_token_notifier: Notify, @@ -441,7 +442,7 @@ impl RoomEventCacheInner { async fn handle_timeline_inner( &self, - mut state: RoomEventCacheStateLockWriteGuard<'_>, + mut state: StateLockWriteGuard<'_, RoomEventCacheState>, timeline: Timeline, ephemeral_events: Vec>, ambiguity_changes: BTreeMap, @@ -802,10 +803,7 @@ mod timed_tests { use tokio::task::yield_now; use super::{ - super::{ - super::TimelineVectorDiffs, lock::Reload as _, - pagination::LoadMoreEventsBackwardsOutcome, - }, + super::{super::TimelineVectorDiffs, pagination::LoadMoreEventsBackwardsOutcome}, RoomEventCache, RoomEventCacheGenericUpdate, RoomEventCacheUpdate, }; use crate::{assert_let_timeout, test_utils::client::MockClientBuilder}; diff --git a/crates/matrix-sdk/src/event_cache/caches/room/state.rs b/crates/matrix-sdk/src/event_cache/caches/room/state.rs index 6bfbe755c..578ad8cb7 100644 --- a/crates/matrix-sdk/src/event_cache/caches/room/state.rs +++ b/crates/matrix-sdk/src/event_cache/caches/room/state.rs @@ -22,10 +22,7 @@ use eyeball_im::VectorDiff; use matrix_sdk_base::{ RoomInfoNotableUpdateReasons, apply_redaction, deserialized_responses::{ThreadSummary, ThreadSummaryStatus}, - event_cache::{ - Event, Gap, - store::{EventCacheStoreLock, EventCacheStoreLockGuard, EventCacheStoreLockState}, - }, + event_cache::{Event, Gap, store::EventCacheStoreLockGuard}, linked_chunk::{ ChunkIdentifierGenerator, LinkedChunkId, OwnedLinkedChunkId, Position, Update, lazy_loader, }, @@ -57,15 +54,14 @@ use super::{ find_event, find_event_relations, find_event_with_relations, load_linked_chunk_metadata, send_updates_to_store, }, + states::{StateLockReadGuard, StateLockWriteGuard}, }, - EventLocation, TimelineVectorDiffs, + EventLocation, event_linked_chunk::EventLinkedChunk, - lock, pagination::SharedPaginationStatus, read_receipts::compute_unread_counts, }, - EventsOrigin, RoomEventCacheGenericUpdate, RoomEventCacheLinkedChunkUpdate, - RoomEventCacheUpdate, RoomEventCacheUpdateSender, sort_positions_descending, + RoomEventCacheLinkedChunkUpdate, RoomEventCacheUpdateSender, sort_positions_descending, }; use crate::room::WeakRoom; @@ -82,9 +78,6 @@ pub struct RoomEventCacheState { /// The user's own user id. pub own_user_id: OwnedUserId, - /// Reference to the underlying backing store. - store: EventCacheStoreLock, - /// The loaded events for the current room, that is, the in-memory /// linked chunk for this room. room_linked_chunk: EventLinkedChunk, @@ -95,7 +88,7 @@ pub struct RoomEventCacheState { /// /// This is used only by the [`RoomEventCacheStateLock::read`] and /// [`RoomEventCacheStateLock::write`] when the state must be reset. - update_sender: RoomEventCacheUpdateSender, + pub update_sender: RoomEventCacheUpdateSender, /// A clone of /// [`super::super::EventCacheInner::linked_chunk_update_sender`]. @@ -119,25 +112,6 @@ pub struct RoomEventCacheState { } impl RoomEventCacheState { - /// Return a read-only reference to the underlying room linked chunk. - pub fn room_linked_chunk(&self) -> &EventLinkedChunk { - &self.room_linked_chunk - } -} - -impl lock::Store for RoomEventCacheState { - fn store(&self) -> &EventCacheStoreLock { - &self.store - } -} - -/// State for a single room's event cache. -/// -/// This contains all the inner mutable states that ought to be updated at -/// the same time. -pub type LockedRoomEventCacheState = lock::StateLock; - -impl LockedRoomEventCacheState { /// Create a new state, or reload it from storage if it's been enabled. /// /// Not all events are going to be loaded. Only a portion of them. The @@ -157,23 +131,10 @@ impl LockedRoomEventCacheState { enabled_thread_support: bool, update_sender: RoomEventCacheUpdateSender, linked_chunk_update_sender: Sender, - store: EventCacheStoreLock, + store_guard: EventCacheStoreLockGuard, pagination_status: SharedObservable, automatic_pagination: Option, ) -> Result { - let store_guard = match store.lock().await? { - // Lock is clean: all good! - EventCacheStoreLockState::Clean(guard) => guard, - - // Lock is dirty, not a problem, it's the first time we are creating this state, no - // need to refresh. - EventCacheStoreLockState::Dirty(guard) => { - EventCacheStoreLockGuard::clear_dirty(&guard); - - guard - } - }; - let linked_chunk_id = LinkedChunkId::Room(&room_id); // Load the full linked chunk's metadata, so as to feed the order tracker. @@ -217,12 +178,11 @@ impl LockedRoomEventCacheState { } }; - Ok(Self::new_inner(RoomEventCacheState { + Ok(RoomEventCacheState { own_user_id, enabled_thread_support, room_id, weak_room, - store, room_linked_chunk: EventLinkedChunk::with_initial_linked_chunk( linked_chunk, full_linked_chunk_metadata, @@ -234,41 +194,16 @@ impl LockedRoomEventCacheState { waited_for_initial_prev_token: false, subscriber_count: Default::default(), automatic_pagination, - })) + }) + } + + /// Return a read-only reference to the underlying room linked chunk. + pub fn room_linked_chunk(&self) -> &EventLinkedChunk { + &self.room_linked_chunk } } -/// The read-lock guard around [`RoomEventCacheState`]. -/// -/// See [`RoomEventCacheStateLock::read`] to acquire it. -pub type RoomEventCacheStateLockReadGuard<'a> = lock::StateLockReadGuard<'a, RoomEventCacheState>; - -/// The write-lock guard around [`RoomEventCacheState`]. -/// -/// See [`RoomEventCacheStateLock::write`] to acquire it. -pub type RoomEventCacheStateLockWriteGuard<'a> = lock::StateLockWriteGuard<'a, RoomEventCacheState>; - -impl<'a> lock::Reload for RoomEventCacheStateLockWriteGuard<'a> { - /// Force to shrink the room, whenever there is subscribers or not. - async fn reload(&mut self) -> Result<(), EventCacheError> { - self.shrink_to_last_chunk().await?; - - let diffs = self.state.room_linked_chunk.updates_as_vector_diffs(); - - // Notify observers about the update. - self.state.update_sender.send( - RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { - diffs, - origin: EventsOrigin::Cache, - }), - Some(RoomEventCacheGenericUpdate { room_id: self.state.room_id.clone() }), - ); - - Ok(()) - } -} - -impl<'a> RoomEventCacheStateLockReadGuard<'a> { +impl<'a> StateLockReadGuard<'a, RoomEventCacheState> { /// Return the subscriber count. pub fn subscriber_count(&self) -> &Arc { &self.state.subscriber_count @@ -328,7 +263,7 @@ impl<'a> RoomEventCacheStateLockReadGuard<'a> { } } -impl<'a> RoomEventCacheStateLockWriteGuard<'a> { +impl<'a> StateLockWriteGuard<'a, RoomEventCacheState> { /// Return a mutable reference to the underlying room linked chunk. pub fn room_linked_chunk_mut(&mut self) -> &mut EventLinkedChunk { &mut self.state.room_linked_chunk @@ -352,6 +287,14 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> { find_event(event_id, &self.room_id, &self.room_linked_chunk, &self.store).await } + /// Force to shrink the room, whenever there is subscribers or not. + #[must_use = "Propagate `VectorDiff` updates via `RoomEventCacheUpdate`"] + pub async fn reload(&mut self) -> Result>, EventCacheError> { + self.shrink_to_last_chunk().await?; + + Ok(self.room_linked_chunk_mut().updates_as_vector_diffs()) + } + /// If storage is enabled, unload all the chunks, then reloads only the /// last one. /// @@ -360,7 +303,7 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> { /// pending diff updates with the result of this function. /// /// Otherwise, returns `None`. - pub async fn shrink_to_last_chunk(&mut self) -> Result<(), EventCacheError> { + async fn shrink_to_last_chunk(&mut self) -> Result<(), EventCacheError> { // Attempt to load the last chunk. let linked_chunk_id = LinkedChunkId::Room(&self.state.room_id); diff --git a/crates/matrix-sdk/src/event_cache/mod.rs b/crates/matrix-sdk/src/event_cache/mod.rs index 746382f5e..2c0cf6478 100644 --- a/crates/matrix-sdk/src/event_cache/mod.rs +++ b/crates/matrix-sdk/src/event_cache/mod.rs @@ -786,7 +786,7 @@ impl EventCacheInner { self.auto_shrink_sender.get().cloned().expect( "we must have called `EventCache::subscribe()` before calling here.", ), - self.store.clone(), + &self.state, self.automatic_pagination.get().cloned(), ) .await?; diff --git a/crates/matrix-sdk/src/event_cache/states/mod.rs b/crates/matrix-sdk/src/event_cache/states/mod.rs index b6b8f3a29..3372f2c56 100644 --- a/crates/matrix-sdk/src/event_cache/states/mod.rs +++ b/crates/matrix-sdk/src/event_cache/states/mod.rs @@ -27,13 +27,19 @@ use matrix_sdk_base::event_cache::store::{ use ruma::{OwnedEventId, OwnedRoomId}; use tokio::sync::{Mutex, RwLock, RwLockMappedWriteGuard, RwLockReadGuard, RwLockWriteGuard}; -use super::{EventCacheError, Result, caches::event_focused::EventFocusedCacheKey}; +use super::{ + EventCacheError, EventsOrigin, Result, + caches::{ + TimelineVectorDiffs, + event_focused::EventFocusedCacheKey, + room::{self, RoomEventCacheState}, + }, +}; -mod selectors; +pub(in super::super) mod selectors; // Temporary types to make the code compiles. Will be removed one after the // other. -pub struct RoomEventCacheState; pub struct ThreadEventCacheState; pub struct PinnedEventsCacheState; pub struct EventFocusedCacheState; @@ -324,7 +330,29 @@ impl<'state> ReloadableStateLockWriteGuard<'state> { } async fn reload(&mut self) -> Result<()> { - todo!() + // Iterate over all states and reload them. + for StateForRoom { room, threads: _, pinned_events: _, event_focused: _ } in + self.state.by_room.values_mut() + { + // Room. + if let Some(room) = room { + let mut room_state = StateLockWriteGuard { + state: StateLockWriteGuardKind::Reference(room), + store: self.store.clone(), + }; + + let updates_as_vector_diffs = room_state.reload().await?; + room_state.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() }), + ); + } + } + + Ok(()) } } diff --git a/crates/matrix-sdk/src/event_cache/states/selectors.rs b/crates/matrix-sdk/src/event_cache/states/selectors.rs index f097dfd56..1d53fa143 100644 --- a/crates/matrix-sdk/src/event_cache/states/selectors.rs +++ b/crates/matrix-sdk/src/event_cache/states/selectors.rs @@ -45,6 +45,12 @@ pub trait CacheState { /// Select a [`RoomEventCacheState`] in [`State`]. pub struct RoomStateSelector(OwnedRoomId); +impl RoomStateSelector { + pub fn new(room_id: OwnedRoomId) -> Self { + Self(room_id) + } +} + impl CacheState for RoomStateSelector { type Item = RoomEventCacheState;