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

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

Browse Source
This commit is contained in:
Ivan Enderlin
2026-06-01 17:17:58 +02:00
parent 7ab448e03c
commit abe32fc2b8
7 changed files with 118 additions and 122 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
crates/matrix-sdk/src/event_cache/caches/aggregator.rs
View File

@@ -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<OwnedEventId, ThreadEventCache>,
room_event_cache: RoomEventCacheStateLockReadGuard<'_>,
room_event_cache: StateLockReadGuard<'_, RoomEventCacheState>,
redaction_rules: &RedactionRules,
) -> Result<HashMap<OwnedEventId, Timeline>> {
let mut new_events_by_thread = HashMap::new();

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

@@ -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::RoomEventCacheLinkedChunkUpdate>,
room_version_rules: RoomVersionRules,
}
@@ -84,7 +86,7 @@ impl Caches {
generic_update_sender: Sender<room::RoomEventCacheGenericUpdate>,
linked_chunk_update_sender: Sender<room::RoomEventCacheLinkedChunkUpdate>,
auto_shrink_sender: mpsc::Sender<OwnedRoomId>,
store: EventCacheStoreLock,
state: &states::StateLock,
automatic_pagination: Option<AutomaticPagination>,
) -> Result<Self> {
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<HashMap<OwnedEventId, thread::ThreadEventCache>>,
thread_locks: Vec<(
thread::OwnedThreadEventCacheStateLockWriteGuard,

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

@@ -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<RoomStateSelector>,
shared_pagination_status: SharedObservable<SharedPaginationStatus>,
auto_shrink_sender: mpsc::Sender<AutoShrinkChannelPayload>,
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<RoomStateSelector> {
&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<RoomStateSelector>,
/// 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<Raw<AnySyncEphemeralRoomEvent>>,
ambiguity_changes: BTreeMap<OwnedEventId, AmbiguityChange>,
@@ -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};

105
crates/matrix-sdk/src/event_cache/caches/room/state.rs
View File

@@ -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<RoomEventCacheState>;
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<RoomEventCacheLinkedChunkUpdate>,
store: EventCacheStoreLock,
store_guard: EventCacheStoreLockGuard,
pagination_status: SharedObservable<SharedPaginationStatus>,
automatic_pagination: Option<AutomaticPagination>,
) -> Result<Self, EventCacheError> {
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<AtomicUsize> {
&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<Vec<VectorDiff<Event>>, 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);

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

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

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

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

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

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