From e9ff0887efc42601758c4deebfbec4ce80aac43f Mon Sep 17 00:00:00 2001 From: Ivan Enderlin Date: Tue, 2 Jun 2026 12:03:16 +0200 Subject: [PATCH] feat(sdk): Migrate `ThreadEventCache` to the new State API. --- .../matrix-sdk/src/event_cache/caches/mod.rs | 3 +- .../src/event_cache/caches/thread/mod.rs | 49 ++++---- .../src/event_cache/caches/thread/state.rs | 118 +++--------------- .../matrix-sdk/src/event_cache/states/mod.rs | 29 ++++- .../src/event_cache/states/selectors.rs | 6 + 5 files changed, 78 insertions(+), 127 deletions(-) diff --git a/crates/matrix-sdk/src/event_cache/caches/mod.rs b/crates/matrix-sdk/src/event_cache/caches/mod.rs index bb4f687c9..ad47a8cc9 100644 --- a/crates/matrix-sdk/src/event_cache/caches/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/mod.rs @@ -204,8 +204,7 @@ impl Caches { room.own_user_id().to_owned(), self.internals.room_version_rules.clone(), room.weak_room().to_owned(), - // self.internals.store.clone(), - todo!(), + &self.internals.state, room.update_sender().generic_update_sender().clone(), self.internals.linked_chunk_update_sender.clone(), ) diff --git a/crates/matrix-sdk/src/event_cache/caches/thread/mod.rs b/crates/matrix-sdk/src/event_cache/caches/thread/mod.rs index c6c709e95..58fd49c02 100644 --- a/crates/matrix-sdk/src/event_cache/caches/thread/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/thread/mod.rs @@ -21,7 +21,7 @@ mod updates; use std::{fmt, sync::Arc}; use matrix_sdk_base::{ - event_cache::{Event, store::EventCacheStoreLock}, + event_cache::Event, sync::{JoinedRoomUpdate, LeftRoomUpdate, Timeline}, }; use ruma::{ @@ -35,14 +35,15 @@ use tokio::sync::{ use tracing::{instrument, trace}; use self::pagination::ThreadPagination; -pub(super) use self::{ - state::{LockedThreadEventCacheState, OwnedThreadEventCacheStateLockWriteGuard}, - updates::ThreadEventCacheUpdateSender, -}; +pub(in super::super) use self::state::ThreadEventCacheState; +pub(super) use self::updates::ThreadEventCacheUpdateSender; #[cfg(feature = "e2e-encryption")] use super::super::redecryptor::ResolvedUtd; use super::{ - super::Result, + super::{ + Result, + states::{CacheStateLock, StateLock, selectors::ThreadStateSelector}, + }, EventsOrigin, TimelineVectorDiffs, room::{RoomEventCacheGenericUpdate, RoomEventCacheLinkedChunkUpdate}, }; @@ -68,7 +69,7 @@ struct ThreadEventCacheInner { weak_room: WeakRoom, /// State for this thread's event cache. - state: LockedThreadEventCacheState, + state: CacheStateLock, /// A notifier that we received a new pagination token. pagination_batch_token_notifier: Notify, @@ -92,32 +93,38 @@ impl ThreadEventCache { own_user_id: OwnedUserId, room_version_rules: RoomVersionRules, weak_room: WeakRoom, - store: EventCacheStoreLock, + state: &StateLock, generic_update_sender: Sender, linked_chunk_update_sender: Sender, ) -> Result { let update_sender = ThreadEventCacheUpdateSender::new(generic_update_sender.clone()); - let state = LockedThreadEventCacheState::new( - room_id.clone(), - thread_id.clone(), - own_user_id, - room_version_rules, - store, - update_sender.clone(), - linked_chunk_update_sender, - ) - .await?; + let cache_state = state + .try_insert_once_with( + ThreadStateSelector::new(room_id.clone(), thread_id.clone()), + |store_guard| { + ThreadEventCacheState::new( + room_id.clone(), + thread_id.clone(), + own_user_id, + room_version_rules, + store_guard, + update_sender.clone(), + linked_chunk_update_sender, + ) + }, + ) + .await?; let timeline_is_not_empty = - state.read().await?.thread_linked_chunk().revents().next().is_some(); + cache_state.read().await?.thread_linked_chunk().revents().next().is_some(); let cache = Self { inner: Arc::new(ThreadEventCacheInner { room_id: room_id.clone(), thread_id, weak_room, - state, + state: cache_state, pagination_batch_token_notifier: Notify::new(), update_sender, }), @@ -162,7 +169,7 @@ impl ThreadEventCache { } /// Return a reference to the state. - pub(in super::super) fn state(&self) -> &LockedThreadEventCacheState { + pub(in super::super) fn state(&self) -> &CacheStateLock { &self.inner.state } diff --git a/crates/matrix-sdk/src/event_cache/caches/thread/state.rs b/crates/matrix-sdk/src/event_cache/caches/thread/state.rs index 6d9bf676e..469c71a8a 100644 --- a/crates/matrix-sdk/src/event_cache/caches/thread/state.rs +++ b/crates/matrix-sdk/src/event_cache/caches/thread/state.rs @@ -16,10 +16,7 @@ use eyeball_im::VectorDiff; use matrix_sdk_base::{ apply_redaction, check_validity_of_replacement_events, deserialized_responses::ThreadSummary, - event_cache::{ - Event, Gap, - store::{EventCacheStoreLock, EventCacheStoreLockGuard, EventCacheStoreLockState}, - }, + event_cache::{Event, Gap, store::EventCacheStoreLockGuard}, linked_chunk::{ ChunkIdentifierGenerator, LinkedChunkId, OwnedLinkedChunkId, Position, Update, lazy_loader, }, @@ -40,17 +37,17 @@ use super::super::super::redecryptor::ResolvedUtd; use super::{ super::{ super::{ - EventCacheError, EventsOrigin, Result, + EventCacheError, Result, deduplicator::{DeduplicationOutcome, filter_duplicate_events}, persistence::{ find_event, find_event_with_relations, load_linked_chunk_metadata, send_updates_to_store, }, + states::{StateLockReadGuard, StateLockWriteGuard}, }, - EventLocation, TimelineVectorDiffs, + EventLocation, event_linked_chunk::{EventLinkedChunk, sort_positions_descending}, - lock, - room::{RoomEventCacheGenericUpdate, RoomEventCacheLinkedChunkUpdate}, + room::RoomEventCacheLinkedChunkUpdate, }, ThreadEventCacheUpdateSender, }; @@ -69,9 +66,6 @@ pub struct ThreadEventCacheState { /// The rules for the version of this room. room_version_rules: RoomVersionRules, - /// Reference to the underlying backing store. - store: EventCacheStoreLock, - /// The linked chunk for this thread. thread_linked_chunk: EventLinkedChunk, @@ -79,7 +73,7 @@ pub struct ThreadEventCacheState { /// /// This is used only by the [`LockedThreadEventCacheState::read`] and /// [`LockedThreadEventCacheState::write`] when the state must be reset. - update_sender: ThreadEventCacheUpdateSender, + pub update_sender: ThreadEventCacheUpdateSender, /// A sender for the globally observable linked chunk updates that happened /// during a sync or a back-pagination. @@ -173,19 +167,7 @@ impl ThreadEventCacheState { } } -impl lock::Store for ThreadEventCacheState { - fn store(&self) -> &EventCacheStoreLock { - &self.store - } -} - -/// State for a single thread's event cache. -/// -/// This contains all the inner mutable states that ought to be updated at -/// the same time. -pub type LockedThreadEventCacheState = lock::StateLock; - -impl LockedThreadEventCacheState { +impl ThreadEventCacheState { /// 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 @@ -201,23 +183,10 @@ impl LockedThreadEventCacheState { thread_id: OwnedEventId, own_user_id: OwnedUserId, room_version_rules: RoomVersionRules, - store: EventCacheStoreLock, + store_guard: EventCacheStoreLockGuard, update_sender: ThreadEventCacheUpdateSender, linked_chunk_update_sender: Sender, ) -> 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::Thread(&room_id, &thread_id); // Load the full linked chunk's metadata, so as to feed the order tracker. @@ -261,12 +230,11 @@ impl LockedThreadEventCacheState { } }; - Ok(Self::new_inner(ThreadEventCacheState { + Ok(ThreadEventCacheState { room_id, thread_id, own_user_id, room_version_rules, - store, thread_linked_chunk: EventLinkedChunk::with_initial_linked_chunk( linked_chunk, full_linked_chunk_metadata, @@ -274,63 +242,11 @@ impl LockedThreadEventCacheState { update_sender, linked_chunk_update_sender, waited_for_initial_prev_token: false, - })) + }) } } -/// The read-lock guard around [`ThreadEventCacheState`]. -/// -/// See [`ThreadEventCacheStateLock::read`] to acquire it. -pub type ThreadEventCacheStateLockReadGuard<'a> = - lock::StateLockReadGuard<'a, ThreadEventCacheState>; - -/// The write-lock guard around [`ThreadEventCacheState`]. -/// -/// See [`ThreadEventCacheStateLock::write`] to acquire it. -pub type ThreadEventCacheStateLockWriteGuard<'a> = - lock::StateLockWriteGuard<'a, ThreadEventCacheState>; - -/// The owned write-lock guard around [`ThreadEventCacheState`]. -pub type OwnedThreadEventCacheStateLockWriteGuard = - lock::OwnedStateLockWriteGuard; - -impl<'a> lock::Reload for ThreadEventCacheStateLockWriteGuard<'a> { - /// Force to shrink the room, whenever there is subscribers or not. - async fn reload(&mut self) -> Result<()> { - self.state.shrink_to_last_chunk(&self.store).await?; - - let diffs = self.state.thread_linked_chunk.updates_as_vector_diffs(); - - if !diffs.is_empty() { - self.state.update_sender.send( - TimelineVectorDiffs { diffs, origin: EventsOrigin::Cache }, - Some(RoomEventCacheGenericUpdate { room_id: self.room_id.to_owned() }), - ); - } - - Ok(()) - } -} - -impl lock::Reload for OwnedThreadEventCacheStateLockWriteGuard { - /// Force to shrink the room, whenever there is subscribers or not. - async fn reload(&mut self) -> Result<()> { - self.state.shrink_to_last_chunk(&self.store).await?; - - let diffs = self.state.thread_linked_chunk.updates_as_vector_diffs(); - - if !diffs.is_empty() { - self.state.update_sender.send( - TimelineVectorDiffs { diffs, origin: EventsOrigin::Cache }, - Some(RoomEventCacheGenericUpdate { room_id: self.room_id.to_owned() }), - ); - } - - Ok(()) - } -} - -impl<'a> ThreadEventCacheStateLockReadGuard<'a> { +impl<'a> StateLockReadGuard<'a, ThreadEventCacheState> { /// Return a read-only reference to the underlying thread linked chunk. pub fn thread_linked_chunk(&self) -> &EventLinkedChunk { &self.state.thread_linked_chunk @@ -433,7 +349,7 @@ impl<'a> ThreadEventCacheStateLockReadGuard<'a> { } } -impl<'a> ThreadEventCacheStateLockWriteGuard<'a> { +impl<'a> StateLockWriteGuard<'a, ThreadEventCacheState> { /// Return a read-only reference to the underlying thread linked chunk. pub fn thread_linked_chunk(&self) -> &EventLinkedChunk { &self.state.thread_linked_chunk @@ -454,6 +370,14 @@ impl<'a> ThreadEventCacheStateLockWriteGuard<'a> { &mut self.state.waited_for_initial_prev_token } + /// Force to shrink the room, whenever there is subscribers or not. + #[must_use = "Propagate `VectorDiff` updates via `TimelineVectorDiffs`"] + pub async fn reload(&mut self) -> Result>> { + self.state.shrink_to_last_chunk(&self.store).await?; + + Ok(self.thread_linked_chunk_mut().updates_as_vector_diffs()) + } + #[must_use = "Propagate `VectorDiff` updates via `TimelineVectorDiffs`"] pub async fn handle_sync( &mut self, @@ -697,9 +621,7 @@ impl<'a> ThreadEventCacheStateLockWriteGuard<'a> { None }) } -} -impl OwnedThreadEventCacheStateLockWriteGuard { /// Reset this data structure as if it were brand new. /// /// Return a single diff update that is a clear of all events; as a diff --git a/crates/matrix-sdk/src/event_cache/states/mod.rs b/crates/matrix-sdk/src/event_cache/states/mod.rs index 3372f2c56..eb4157544 100644 --- a/crates/matrix-sdk/src/event_cache/states/mod.rs +++ b/crates/matrix-sdk/src/event_cache/states/mod.rs @@ -33,6 +33,7 @@ use super::{ TimelineVectorDiffs, event_focused::EventFocusedCacheKey, room::{self, RoomEventCacheState}, + thread::ThreadEventCacheState, }, }; @@ -40,7 +41,6 @@ pub(in super::super) mod selectors; // Temporary types to make the code compiles. Will be removed one after the // other. -pub struct ThreadEventCacheState; pub struct PinnedEventsCacheState; pub struct EventFocusedCacheState; @@ -331,13 +331,13 @@ impl<'state> ReloadableStateLockWriteGuard<'state> { async fn reload(&mut self) -> Result<()> { // Iterate over all states and reload them. - for StateForRoom { room, threads: _, pinned_events: _, event_focused: _ } in - self.state.by_room.values_mut() + for (room_id, StateForRoom { room, threads, pinned_events: _, event_focused: _ }) in + self.state.by_room.iter_mut() { // Room. - if let Some(room) = room { + if let Some(room_state) = room { let mut room_state = StateLockWriteGuard { - state: StateLockWriteGuardKind::Reference(room), + state: StateLockWriteGuardKind::Reference(room_state), store: self.store.clone(), }; @@ -347,7 +347,24 @@ impl<'state> ReloadableStateLockWriteGuard<'state> { diffs: updates_as_vector_diffs, origin: EventsOrigin::Cache, }), - Some(room::RoomEventCacheGenericUpdate { room_id: room_state.room_id.clone() }), + Some(room::RoomEventCacheGenericUpdate { room_id: room_id.clone() }), + ); + } + + // Threads. + for thread_state in threads.values_mut() { + let mut thread_state = StateLockWriteGuard { + state: StateLockWriteGuardKind::Reference(thread_state), + store: self.store.clone(), + }; + + let updates_as_vector_diffs = thread_state.reload().await?; + thread_state.update_sender.send( + TimelineVectorDiffs { + diffs: updates_as_vector_diffs, + origin: EventsOrigin::Cache, + }, + Some(room::RoomEventCacheGenericUpdate { room_id: room_id.clone() }), ); } } diff --git a/crates/matrix-sdk/src/event_cache/states/selectors.rs b/crates/matrix-sdk/src/event_cache/states/selectors.rs index 1d53fa143..a42dcbc51 100644 --- a/crates/matrix-sdk/src/event_cache/states/selectors.rs +++ b/crates/matrix-sdk/src/event_cache/states/selectors.rs @@ -85,6 +85,12 @@ impl From<&RoomStateSelector> for EventCacheError { /// Select a [`ThreadEventCacheState`] in [`State`]. pub struct ThreadStateSelector(OwnedRoomId, OwnedEventId); +impl ThreadStateSelector { + pub fn new(room_id: OwnedRoomId, thread_id: OwnedEventId) -> Self { + Self(room_id, thread_id) + } +} + impl CacheState for ThreadStateSelector { type Item = ThreadEventCacheState;