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

feat(sdk): Migrate ThreadEventCache to the new State API.

Browse Source
This commit is contained in:
Ivan Enderlin
2026-06-02 12:03:16 +02:00
parent b1808eceb3
commit e9ff0887ef
5 changed files with 78 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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(),
)

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

@@ -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<ThreadStateSelector>,
/// 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<RoomEventCacheGenericUpdate>,
linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
) -> Result<Self> {
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<ThreadStateSelector> {
&self.inner.state
}

118
crates/matrix-sdk/src/event_cache/caches/thread/state.rs
View File

@@ -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<ThreadEventCacheState>;
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<RoomEventCacheLinkedChunkUpdate>,
) -> Result<Self> {
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<ThreadEventCacheState>;
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<Vec<VectorDiff<Event>>> {
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

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

@@ -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() }),
);
}
}

6
crates/matrix-sdk/src/event_cache/states/selectors.rs
View File

@@ -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;