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

refactor(sdk): Move threads from RoomEventCache to Caches.

Browse Source 
This patch starts flattening caches by starting the extract thread
caches from the room cache.

This patch comments some methods or adds a `todo!()` to make the code
compile. It's the price for smaller commits.

The thread cache gains new methods like `handle_joined_room_update` or
`handle_joined_left_update`. It mimics the flow in the room cache.

A new `caches::aggregator` module is created to introduce “pipelines”:
a `Timeline` is created for each cache. It lives in front of the room
cache, and basically reverses the “post process” flow to a “pre process”
flow. More commits will improve it.

`ResetCaches` is now also responsible to clear the threads in addition
to the room (previously, the room cache was responsible of that).
This commit is contained in:
Ivan Enderlin
2026-04-24 15:16:47 +02:00
parent 90091402a8
commit 30353d3b4e
7 changed files with 355 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
crates/matrix-sdk/src/event_cache/caches/aggregator.rs Normal file
View File

@@ -0,0 +1,75 @@
// Copyright 2026 The Matrix.org Foundation C.I.C.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::collections::HashMap;
use matrix_sdk_base::{
serde_helpers::{extract_edit_target, extract_thread_root},
sync::Timeline,
};
use ruma::OwnedEventId;
use super::{super::Result, room::RoomEventCacheStateLockReadGuard, thread::ThreadEventCache};
pub fn aggregate_timeline_for_room(timeline: Timeline) -> Timeline {
timeline
}
pub async fn aggregate_timeline_for_threads(
timeline: &Timeline,
existing_threads: &HashMap<OwnedEventId, ThreadEventCache>,
room_event_cache: RoomEventCacheStateLockReadGuard<'_>,
) -> Result<HashMap<OwnedEventId, Timeline>> {
let mut new_events_by_thread = HashMap::new();
let default_timeline = || Timeline {
limited: timeline.limited,
prev_batch: timeline.prev_batch.clone(),
events: Vec::new(),
};
for event in &timeline.events {
// This event is part of a thread.
if let Some(thread_root) = extract_thread_root(event.raw()) {
new_events_by_thread
.entry(thread_root)
.or_insert_with(default_timeline)
.events
.push(event.clone());
}
// This event is the root of a thread.
else if let Some(event_id) = event.event_id()
&& existing_threads.contains_key(&event_id)
{
new_events_by_thread
.entry(event_id)
.or_insert_with(default_timeline)
.events
.push(event.clone());
}
// This event is an edit that may apply to a thread..
if let Some(edit_target) = extract_edit_target(event.raw()) {
// This event is known and part of a thread.
if let Some((_location, edit_target_event)) =
room_event_cache.find_event(&edit_target).await?
&& let Some(thread_root) = extract_thread_root(edit_target_event.raw())
{
new_events_by_thread.entry(thread_root).or_insert_with(default_timeline);
}
}
}
Ok(new_events_by_thread)
}

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

@@ -12,6 +12,8 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::{collections::HashMap, ops::Deref, sync::Arc};
use eyeball::SharedObservable;
use eyeball_im::VectorDiff;
use matrix_sdk_base::{
@@ -20,12 +22,13 @@ use matrix_sdk_base::{
linked_chunk::Position,
sync::{JoinedRoomUpdate, LeftRoomUpdate},
};
use ruma::{OwnedRoomId, RoomId};
use tokio::sync::{broadcast::Sender, mpsc};
use ruma::{OwnedEventId, OwnedRoomId, RoomId};
use tokio::sync::{OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock, broadcast::Sender, mpsc};
use super::{EventCacheError, EventsOrigin, Result, automatic_pagination::AutomaticPagination};
use crate::{client::WeakClient, room::WeakRoom};
mod aggregator;
pub mod event_focused;
pub mod event_linked_chunk;
pub(super) mod lock;
@@ -39,6 +42,14 @@ pub mod thread;
#[derive(Debug)]
pub(super) struct Caches {
pub room: room::RoomEventCache,
pub threads: Arc<RwLock<HashMap<OwnedEventId, thread::ThreadEventCache>>>,
internals: CachesInternals,
}
#[derive(Debug)]
struct CachesInternals {
store: EventCacheStoreLock,
linked_chunk_update_sender: Sender<room::RoomEventCacheLinkedChunkUpdate>,
}
impl Caches {
@@ -76,14 +87,14 @@ impl Caches {
client.user_id().expect("the user must be logged in, at this point").to_owned();
let room_state = room::LockedRoomEventCacheState::new(
own_user_id,
own_user_id.clone(),
room_id.to_owned(),
weak_room.clone(),
room_version_rules,
enabled_thread_support,
update_sender.clone(),
linked_chunk_update_sender,
store,
linked_chunk_update_sender.clone(),
store.clone(),
pagination_status.clone(),
automatic_pagination,
)
@@ -95,6 +106,7 @@ impl Caches {
let room_event_cache = room::RoomEventCache::new(
room_id.to_owned(),
weak_room,
own_user_id,
room_state,
pagination_status,
auto_shrink_sender,
@@ -108,23 +120,135 @@ impl Caches {
.send(room::RoomEventCacheGenericUpdate { room_id: room_id.to_owned() });
}
Ok(Self { room: room_event_cache })
Ok(Self {
room: room_event_cache,
threads: Arc::new(RwLock::new(HashMap::new())),
internals: CachesInternals { store, linked_chunk_update_sender },
})
}
/// Get the [`RoomEventCache`].
///
/// [`RoomEventCache`]: room::RoomEventCache
pub async fn room(&self) -> &room::RoomEventCache {
&self.room
}
/// Get or create a [`ThreadEventCache`].
///
/// Note: it is impossible to know if `thread_id` represents a valid thread
/// identifier. It means it's possible to create a [`ThreadEventCache`] for
/// an event that is not a thread root.
///
/// [`ThreadEventCache`]: thread::ThreadEventCache
pub async fn thread(
&self,
thread_id: OwnedEventId,
) -> Result<
OwnedRwLockReadGuard<
HashMap<OwnedEventId, thread::ThreadEventCache>,
thread::ThreadEventCache,
>,
> {
Ok(
match OwnedRwLockWriteGuard::try_downgrade_map(
self.threads.clone().write_owned().await,
|threads| threads.get(&thread_id),
) {
// Thread exists.
Ok(locked_cache) => locked_cache,
// Thread does not exist, let's create it.
Err(mut threads) => {
let room = &self.room;
let cache = thread::ThreadEventCache::new(
room.room_id().to_owned(),
thread_id.clone(),
room.own_user_id().to_owned(),
room.weak_room().to_owned(),
self.internals.store.clone(),
room.update_sender().generic_update_sender().clone(),
self.internals.linked_chunk_update_sender.clone(),
)
.await?;
threads.insert(thread_id.clone(), cache);
OwnedRwLockWriteGuard::downgrade_map(threads, |threads| {
threads.get(&thread_id).unwrap()
})
}
},
)
}
/// Update all the event caches with a [`JoinedRoomUpdate`].
pub(super) async fn handle_joined_room_update(&self, updates: JoinedRoomUpdate) -> Result<()> {
let Self { room } = &self;
let Self { room, threads, internals: _ } = &self;
room.handle_joined_room_update(updates).await?;
// Room.
{
let mut updates = updates.clone();
updates.timeline = aggregator::aggregate_timeline_for_room(updates.timeline);
room.handle_joined_room_update(updates).await?;
}
// Threads.
{
let mut updates = updates.clone();
updates.account_data.clear();
updates.ambiguity_changes.clear();
let timeline_for_threads = aggregator::aggregate_timeline_for_threads(
&updates.timeline,
threads.read().await.deref(),
room.state().read().await?,
)
.await?;
for (thread_id, timeline) in timeline_for_threads {
let mut updates = updates.clone();
updates.timeline = timeline;
self.thread(thread_id).await?.handle_joined_room_update(updates).await?;
}
}
Ok(())
}
/// Update all the event caches with a [`LeftRoomUpdate`].
pub(super) async fn handle_left_room_update(&self, updates: LeftRoomUpdate) -> Result<()> {
let Self { room } = &self;
let Self { room, threads, internals: _ } = &self;
room.handle_left_room_update(updates).await?;
// Room.
{
let mut updates = updates.clone();
updates.timeline = aggregator::aggregate_timeline_for_room(updates.timeline);
room.handle_left_room_update(updates).await?;
}
// Threads.
{
let mut updates = updates.clone();
updates.account_data.clear();
updates.ambiguity_changes.clear();
let timeline_for_threads = aggregator::aggregate_timeline_for_threads(
&updates.timeline,
threads.read().await.deref(),
room.state().read().await?,
)
.await?;
for (thread_id, timeline) in timeline_for_threads {
let mut updates = updates.clone();
updates.timeline = timeline;
self.thread(thread_id).await?.handle_left_room_update(updates).await?;
}
}
Ok(())
}
@@ -157,15 +281,33 @@ 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: (&'c room::RoomEventCache, room::RoomEventCacheStateLockWriteGuard<'c>),
room_lock: (room::RoomEventCacheStateLockWriteGuard<'c>, room::RoomEventCacheUpdateSender),
threads_lock: OwnedRwLockWriteGuard<HashMap<OwnedEventId, thread::ThreadEventCache>>,
thread_locks: Vec<(
thread::OwnedThreadEventCacheStateLockWriteGuard,
thread::ThreadEventCacheUpdateSender,
)>,
}
impl<'c> ResetCaches<'c> {
/// Create a new [`ResetCaches`].
///
/// It can fail if acquiring an exclusive lock fails.
async fn new(Caches { room }: &'c mut Caches) -> Result<Self> {
Ok(Self { room_lock: (room, room.state().write().await?) })
async fn new(Caches { room, threads, internals: _ }: &'c mut Caches) -> Result<Self> {
// Acquire an exclusive access to the state of the room.
let room_lock = (room.state().write().await?, room.update_sender().clone());
// Acquire an exclusive access to the threads.
// Then, for each thread, acquire an exclusive access to its state.
let threads_lock = threads.clone().write_owned().await;
let mut thread_locks = Vec::new();
for thread in threads_lock.values() {
thread_locks
.push((thread.state().write_owned().await?, thread.update_sender().clone()));
}
Ok(Self { room_lock, threads_lock, thread_locks })
}
/// Reset all the event caches, and broadcast the [`TimelineVectorDiffs`].
@@ -175,19 +317,41 @@ impl<'c> ResetCaches<'c> {
///
/// It can fail if resetting an event cache fails.
pub async fn reset_all(self) -> Result<()> {
let Self { room_lock: (room, mut room_state) } = self;
let Self { room_lock, threads_lock, thread_locks } = self;
{
let (mut room_state, room_update_sender) = room_lock;
let updates_as_vector_diffs = room_state.reset().await?;
room.update_sender().send(
room_update_sender.send(
room::RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
diffs: updates_as_vector_diffs,
origin: EventsOrigin::Cache,
}),
Some(room::RoomEventCacheGenericUpdate { room_id: room.room_id().to_owned() }),
Some(room::RoomEventCacheGenericUpdate { room_id: room_state.room_id.clone() }),
);
}
{
for thread_lock in thread_locks {
let (mut thread_state, thread_update_sender) = thread_lock;
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 acces over the threads.
drop(threads_lock);
}
Ok(())
}
}

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

@@ -37,7 +37,9 @@ use ruma::{
use tokio::sync::{Notify, broadcast::Receiver, mpsc};
use tracing::{instrument, trace, warn};
pub(super) use self::state::{LockedRoomEventCacheState, RoomEventCacheStateLockWriteGuard};
pub(super) use self::state::{
LockedRoomEventCacheState, RoomEventCacheStateLockReadGuard, RoomEventCacheStateLockWriteGuard,
};
pub use self::{
subscriber::RoomEventCacheSubscriber,
updates::{
@@ -79,20 +81,23 @@ impl RoomEventCache {
pub(super) fn new(
room_id: OwnedRoomId,
weak_room: WeakRoom,
own_user_id: OwnedUserId,
state: LockedRoomEventCacheState,
shared_pagination_status: SharedObservable<SharedPaginationStatus>,
auto_shrink_sender: mpsc::Sender<AutoShrinkChannelPayload>,
update_sender: RoomEventCacheUpdateSender,
) -> Self {
Self {
inner: Arc::new(RoomEventCacheInner::new(
inner: Arc::new(RoomEventCacheInner {
room_id,
weak_room,
own_user_id,
state,
shared_pagination_status,
auto_shrink_sender,
update_sender,
)),
pagination_batch_token_notifier: Notify::new(),
auto_shrink_sender,
shared_pagination_status,
}),
}
}
@@ -101,6 +106,16 @@ impl RoomEventCache {
&self.inner.room_id
}
/// Get the owner of this [`RoomEventCache`].
pub(super) fn own_user_id(&self) -> &OwnedUserId {
&self.inner.own_user_id
}
/// Get the weak room of this [`RoomEventCache`].
pub(super) fn weak_room(&self) -> &WeakRoom {
&self.inner.weak_room
}
/// Read all current events.
///
/// Use [`RoomEventCache::subscribe`] to get all current events, plus a
@@ -443,6 +458,9 @@ pub(super) struct RoomEventCacheInner {
weak_room: WeakRoom,
/// The user's own user id.
own_user_id: OwnedUserId,
/// State for this room's event cache.
state: LockedRoomEventCacheState,
@@ -462,27 +480,6 @@ pub(super) struct RoomEventCacheInner {
}
impl RoomEventCacheInner {
/// Creates a new cache for a room, and subscribes to room updates, so as
/// to handle new timeline events.
fn new(
room_id: OwnedRoomId,
weak_room: WeakRoom,
state: LockedRoomEventCacheState,
shared_pagination_status: SharedObservable<SharedPaginationStatus>,
auto_shrink_sender: mpsc::Sender<AutoShrinkChannelPayload>,
update_sender: RoomEventCacheUpdateSender,
) -> Self {
Self {
room_id,
weak_room,
state,
update_sender,
pagination_batch_token_notifier: Notify::new(),
auto_shrink_sender,
shared_pagination_status,
}
}
fn handle_account_data(&self, account_data: Vec<Raw<AnyRoomAccountDataEvent>>) {
if account_data.is_empty() {
return;

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

@@ -13,8 +13,7 @@
// limitations under the License.
use std::{
collections::{BTreeMap, HashMap, HashSet, hash_map::Entry},
ops::DerefMut,
collections::{BTreeMap, HashMap, HashSet},
sync::{
Arc, OnceLock,
atomic::{AtomicUsize, Ordering},
@@ -785,15 +784,6 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
async fn reset_internal(&mut self) -> Result<(), EventCacheError> {
self.state.room_linked_chunk.reset();
// No need to update the thread summaries: the room events are
// gone because of the reset of `room_linked_chunk`.
//
// Clear the threads.
for thread in self.state.threads.values_mut() {
thread.clear().await?;
}
self.propagate_changes().await?;
// Reset the pagination state too: pretend we never waited for the initial
@@ -1059,6 +1049,7 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
Ok(())
}
/*
pub(super) async fn get_or_reload_thread(
&mut self,
root_event_id: OwnedEventId,
@@ -1093,6 +1084,7 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
Entry::Occupied(entry) => Ok(entry.into_mut()),
}
}
*/
#[instrument(skip_all)]
async fn update_threads(

2
crates/matrix-sdk/src/event_cache/caches/room/updates.rs
View File

@@ -139,7 +139,7 @@ impl RoomEventCacheUpdateSender {
}
/// Get the generic update sender.
pub(super) fn generic_update_sender(&self) -> &Sender<RoomEventCacheGenericUpdate> {
pub(in super::super) fn generic_update_sender(&self) -> &Sender<RoomEventCacheGenericUpdate> {
&self.generic_sender
}

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

@@ -20,16 +20,22 @@ mod updates;
use std::{fmt, sync::Arc};
use matrix_sdk_base::event_cache::{Event, store::EventCacheStoreLock};
use matrix_sdk_base::{
event_cache::{Event, store::EventCacheStoreLock},
sync::{JoinedRoomUpdate, LeftRoomUpdate, Timeline},
};
use ruma::{EventId, OwnedEventId, OwnedRoomId, OwnedUserId};
use tokio::sync::{
Notify,
broadcast::{Receiver, Sender},
};
use tracing::{error, trace};
use tracing::{error, instrument, trace};
pub(super) use self::state::LockedThreadEventCacheState;
use self::{pagination::ThreadPagination, updates::ThreadEventCacheUpdateSender};
use self::pagination::ThreadPagination;
pub(super) use self::{
state::{LockedThreadEventCacheState, OwnedThreadEventCacheStateLockWriteGuard},
updates::ThreadEventCacheUpdateSender,
};
use super::{
super::Result,
EventsOrigin, TimelineVectorDiffs,
@@ -39,7 +45,7 @@ use super::{
use crate::room::WeakRoom;
/// All the information related to a single thread.
pub(super) struct ThreadEventCache {
pub struct ThreadEventCache {
inner: Arc<ThreadEventCacheInner>,
}
@@ -135,13 +141,53 @@ impl ThreadEventCache {
ThreadPagination::new(self.inner.clone())
}
/// Clear a thread.
pub async fn clear(&mut self) -> Result<()> {
let updates_as_vector_diffs = self.inner.state.write().await?.reset().await?;
/// Return a reference to the state.
pub(in super::super) fn state(&self) -> &LockedThreadEventCacheState {
&self.inner.state
}
if !updates_as_vector_diffs.is_empty() {
/// Get a reference to the [`RoomEventCacheUpdateSender`].
pub(in super::super) fn update_sender(&self) -> &ThreadEventCacheUpdateSender {
&self.inner.update_sender
}
/// Handle a [`JoinedRoomUpdate`].
#[instrument(skip_all, fields(room_id = %self.inner.room_id, thread_root = %self.inner.thread_id))]
pub(super) async fn handle_joined_room_update(&self, updates: JoinedRoomUpdate) -> Result<()> {
self.handle_timeline(updates.timeline).await?;
Ok(())
}
/// Handle a [`LeftRoomUpdate`].
#[instrument(skip_all, fields(room_id = %self.inner.room_id, thread_root = %self.inner.thread_id))]
pub(super) async fn handle_left_room_update(&self, updates: LeftRoomUpdate) -> Result<()> {
self.handle_timeline(updates.timeline).await?;
Ok(())
}
/// Handle a [`Timeline`], i.e. new events received by a sync for this
/// thread.
async fn handle_timeline(&self, timeline: Timeline) -> Result<()> {
if timeline.events.is_empty() && timeline.prev_batch.is_none() {
return Ok(());
}
trace!("adding new events");
let (stored_prev_batch_token, timeline_event_diffs) =
self.inner.state.write().await?.handle_sync(timeline).await?;
// Now that all events have been added, we can trigger the
// `pagination_token_notifier`.
if stored_prev_batch_token {
self.inner.pagination_batch_token_notifier.notify_one();
}
if !timeline_event_diffs.is_empty() {
self.inner.update_sender.send(
TimelineVectorDiffs { diffs: updates_as_vector_diffs, origin: EventsOrigin::Cache },
TimelineVectorDiffs { diffs: timeline_event_diffs, origin: EventsOrigin::Sync },
// This function is part of the `RoomEventCache` flow. The generic update is
// handled by it.
None,
@@ -158,6 +204,8 @@ impl ThreadEventCache {
events: Vec<Event>,
prev_batch_token: &Option<String>,
) -> Result<()> {
todo!()
/*
if events.is_empty() {
return Ok(());
}
@@ -183,6 +231,7 @@ impl ThreadEventCache {
}
Ok(())
*/
}
/// Replaces a single event, be it saved in memory or in the store.

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

@@ -21,6 +21,7 @@ use matrix_sdk_base::{
linked_chunk::{
ChunkIdentifierGenerator, LinkedChunkId, OwnedLinkedChunkId, Position, Update, lazy_loader,
},
sync::Timeline,
};
use matrix_sdk_common::executor::spawn;
use ruma::{EventId, OwnedEventId, OwnedRoomId, OwnedUserId};
@@ -343,9 +344,10 @@ impl<'a> ThreadEventCacheStateLockWriteGuard<'a> {
#[must_use = "Propagate `VectorDiff` updates via `TimelineVectorDiffs`"]
pub async fn handle_sync(
&mut self,
events: Vec<Event>,
prev_batch_token: &Option<String>,
timeline: Timeline,
) -> Result<(bool, Vec<VectorDiff<Event>>)> {
let prev_batch_token = &timeline.prev_batch;
let DeduplicationOutcome {
all_events: events,
in_memory_duplicated_event_ids,
@@ -356,7 +358,7 @@ impl<'a> ThreadEventCacheStateLockWriteGuard<'a> {
&self.store,
LinkedChunkId::Thread(&self.state.room_id, &self.state.thread_id),
&self.state.thread_linked_chunk,
events,
timeline.events,
)
.await?;
@@ -387,6 +389,15 @@ impl<'a> ThreadEventCacheStateLockWriteGuard<'a> {
self.state.propagate_changes(&self.store).await?;
if timeline.limited && has_new_gap {
// If there was a previous batch token for a limited timeline, unload the chunks
// so it only contains the last one; otherwise, there might be a
// valid gap in between, and observers may not render it (yet).
//
// We must do this *after* persisting these events to storage.
self.state.shrink_to_last_chunk(&self.store).await?;
}
let timeline_event_diffs = self.state.thread_linked_chunk.updates_as_vector_diffs();
Ok((has_new_gap, timeline_event_diffs))