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 EventFocusedCache to the new State API.

Browse Source
This commit is contained in:
Ivan Enderlin
2026-06-02 15:53:43 +02:00
parent aa4816b710
commit f94ee60b70
7 changed files with 137 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
crates/matrix-sdk-ui/src/timeline/controller/mod.rs
View File

@@ -1384,13 +1384,13 @@ impl TimelineController {
event_cache,
..
} => {
let (events, receiver) = event_cache.subscribe().await;
let (events, receiver) = event_cache.subscribe().await?;
let has_events = !events.is_empty();
// Ask the cache for the thread root, if it managed to extract one or decided
// that the target event was the thread root.
if let Some(thread_root) = event_cache.thread_root().await {
if let Some(thread_root) = event_cache.thread_root().await? {
focus_thread_root.get_or_init(|| thread_root);
}

5
crates/matrix-sdk-ui/src/timeline/tasks.rs
View File

@@ -112,7 +112,10 @@ pub(in crate::timeline) async fn event_focused_task(
// The updates might have lagged, but the room event cache might have
// events, so retrieve them and add them back again to the timeline,
// after clearing it.
let (initial_events, _) = event_cache.subscribe().await;
let Ok((initial_events, _)) = event_cache.subscribe().await else {
error!("Failed to subscribe to the event-focused cache");
break;
};
timeline_controller
.replace_with_initial_remote_events(initial_events, RemoteEventOrigin::Cache)

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

@@ -39,20 +39,21 @@ use matrix_sdk_base::{
};
use matrix_sdk_common::{linked_chunk::ChunkIdentifier, serde_helpers::extract_thread_root};
use ruma::{OwnedEventId, UInt, api::Direction};
use tokio::sync::{
RwLock,
broadcast::{Receiver, Sender},
};
use tokio::sync::broadcast::{Receiver, Sender};
use tracing::{instrument, trace};
#[cfg(feature = "e2e-encryption")]
use crate::event_cache::redecryptor::ResolvedUtd;
use super::super::redecryptor::ResolvedUtd;
use super::{
super::{
EventCacheError, EventsOrigin, Result, RoomEventCacheLinkedChunkUpdate,
states::{CacheStateLock, StateLock, selectors::EventFocusedStateSelector},
},
TimelineVectorDiffs,
event_linked_chunk::EventLinkedChunk,
};
use crate::{
Room,
event_cache::{
EventCacheError, EventsOrigin, Result, RoomEventCacheLinkedChunkUpdate,
caches::{TimelineVectorDiffs, event_linked_chunk::EventLinkedChunk},
},
paginators::{PaginationResult, Paginator, StartFromResult, thread::PaginableThread},
room::{IncludeRelations, MessagesOptions, RelationsOptions, WeakRoom},
};
@@ -91,7 +92,7 @@ pub(crate) enum EventFocusedPaginationMode {
},
}
pub(super) struct EventFocusedCacheState {
pub struct EventFocusedCacheState {
/// The room owning this event-focused cache.
room: WeakRoom,
@@ -104,8 +105,16 @@ pub(super) struct EventFocusedCacheState {
/// The linked chunk for this event-focused cache.
chunk: EventLinkedChunk,
/// The `num_context_events` given to [`Self::start_from`].
///
/// This is useful for [`Self::reload`] to load the same amount of events.
initial_num_context_events: u16,
/// The thread mode.
thread_mode: EventFocusThreadMode,
/// A sender of timeline updates.
sender: EventFocusedCacheUpdateSender,
pub update_sender: EventFocusedCacheUpdateSender,
/// A sender for globally observable linked chunk updates.
linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
@@ -128,7 +137,36 @@ impl EventFocusedCacheState {
num_context_events: u16,
thread_mode: EventFocusThreadMode,
) -> Result<StartFromResult> {
self.initial_num_context_events = num_context_events;
self.thread_mode = thread_mode;
let result = self.reload_impl().await?;
// Empty the updates_as_vector_diffs(), since it's impossible for an observer to
// have subscribed to this cache yet, since this code is part of the constructor
// flow.
//
// If we didn't empty those, such initial updates would be duplicated, since the
// subscriber would get the full initial list of events as diffs and as a set of
// initial events.
let _ = self.chunk.updates_as_vector_diffs();
Ok(result)
}
#[must_use = "Propagate `VectorDiff` updates via `TimelineVectorDiffs`"]
pub async fn reload(&mut self) -> Result<Vec<VectorDiff<Event>>> {
let _ = self.reload_impl().await?;
Ok(self.chunk.updates_as_vector_diffs())
}
/// Replace existing events, then load and store fresh events from
/// `/context`.
async fn reload_impl(&mut self) -> Result<StartFromResult> {
let room = self.room.get().ok_or(EventCacheError::ClientDropped)?;
let num_context_events = self.initial_num_context_events;
let thread_mode = self.thread_mode;
trace!(num_context_events, "fetching event with context via /context");
@@ -237,15 +275,6 @@ impl EventFocusedCacheState {
self.propagate_changes();
// Empty the updates_as_vector_diffs(), since it's impossible for an observer to
// have subscribed to this cache yet, since this code is part of the constructor
// flow.
//
// If we didn't empty those, such initial updates would be duplicated, since the
// subscriber would get the full initial list of events as diffs and as a set of
// initial events.
let _ = self.chunk.updates_as_vector_diffs();
Ok(result)
}
@@ -256,6 +285,9 @@ impl EventFocusedCacheState {
prev_gap_token: Option<String>,
next_gap_token: Option<String>,
) {
// Clear all existing events as we are about to insert initial events.
self.chunk.reset();
// Insert backward gap at the back if we have a token, and the events
// themselves.
self.chunk
@@ -286,7 +318,7 @@ impl EventFocusedCacheState {
fn notify_subscribers(&mut self, origin: EventsOrigin) {
let diffs = self.chunk.updates_as_vector_diffs();
if !diffs.is_empty() {
let _ = self.sender.send(TimelineVectorDiffs { diffs, origin });
let _ = self.update_sender.send(TimelineVectorDiffs { diffs, origin });
}
}
@@ -526,7 +558,7 @@ impl EventFocusedCacheState {
/// A cache for an event-focused timeline.
///
/// This represents a timeline centered around a specific event (e.g., from a
/// This represents a timeline centred around a specific event (e.g., from a
/// permalink), supporting both forward and backward pagination. The focused
/// event may be part of a thread, in which case pagination will use the
/// `/relations` API instead of `/messages`.
@@ -540,56 +572,68 @@ impl EventFocusedCacheState {
/// This is a shallow data structure, and can be cloned cheaply.
#[derive(Clone)]
pub struct EventFocusedCache {
inner: Arc<RwLock<EventFocusedCacheState>>,
inner: Arc<CacheStateLock<EventFocusedStateSelector>>,
}
impl EventFocusedCache {
/// Create a new empty event-focused cache.
pub(super) fn new(
pub(super) async fn new(
room: WeakRoom,
focused_event_id: OwnedEventId,
key: EventFocusedCacheKey,
state: &StateLock,
linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
) -> Self {
Self {
inner: Arc::new(RwLock::new(EventFocusedCacheState {
room,
focused_event_id,
pagination_mode: EventFocusedPaginationMode::Room { hide_thread_events: false },
chunk: EventLinkedChunk::new(),
sender: Sender::new(32),
linked_chunk_update_sender,
})),
}
) -> Result<Self> {
let cache_state = state
.try_insert_once_with(
EventFocusedStateSelector::new(room.room_id().to_owned(), key.clone()),
|_store_guard| async {
Ok(EventFocusedCacheState {
room,
focused_event_id: key.focused_event_id,
pagination_mode: EventFocusedPaginationMode::Room {
hide_thread_events: false,
},
chunk: EventLinkedChunk::new(),
initial_num_context_events: 0, // dummy value
thread_mode: EventFocusThreadMode::Automatic, // dummy value
update_sender: Sender::new(32),
linked_chunk_update_sender,
})
},
)
.await?;
Ok(Self { inner: Arc::new(cache_state) })
}
/// Return a reference to the state.
pub(super) fn state(&self) -> &Arc<RwLock<EventFocusedCacheState>> {
pub(super) fn state(&self) -> &CacheStateLock<EventFocusedStateSelector> {
&self.inner
}
/// Get a reference to the _update sender_.
pub(super) async fn update_sender(&self) -> EventFocusedCacheUpdateSender {
self.inner.read().await.sender.clone()
pub(super) async fn update_sender(&self) -> Result<EventFocusedCacheUpdateSender> {
Ok(self.inner.read().await?.update_sender.clone())
}
/// Subscribe to updates from this event-focused timeline.
pub async fn subscribe(&self) -> (Vec<Event>, Receiver<TimelineVectorDiffs>) {
let inner = self.inner.read().await;
pub async fn subscribe(&self) -> Result<(Vec<Event>, Receiver<TimelineVectorDiffs>)> {
let inner = self.inner.read().await?;
let events = inner.chunk.events().map(|(_position, item)| item.clone()).collect();
let recv = inner.sender.subscribe();
(events, recv)
let recv = inner.update_sender.subscribe();
Ok((events, recv))
}
/// Check if we've hit the start of the timeline (no more backward
/// pagination possible).
pub async fn hit_timeline_start(&self) -> bool {
self.inner.read().await.first_chunk_as_gap().is_none()
pub async fn hit_timeline_start(&self) -> Result<bool> {
Ok(self.inner.read().await?.first_chunk_as_gap().is_none())
}
/// Check if we've hit the end of the timeline (no more forward pagination
/// possible).
pub async fn hit_timeline_end(&self) -> bool {
self.inner.read().await.last_chunk_as_gap().is_none()
pub async fn hit_timeline_end(&self) -> Result<bool> {
Ok(self.inner.read().await?.last_chunk_as_gap().is_none())
}
/// Start the event-focused timeline from the focused event, fetching
@@ -599,39 +643,42 @@ impl EventFocusedCache {
num_context_events: u16,
thread_mode: EventFocusThreadMode,
) -> Result<StartFromResult> {
self.inner.write().await.start_from(num_context_events, thread_mode).await
self.inner.write().await?.start_from(num_context_events, thread_mode).await
}
/// Paginate backwards in this event-focused timeline, be it room or thread
/// pagination depending on the mode.
pub async fn paginate_backwards(&self, num_events: u16) -> Result<PaginationResult> {
self.inner.write().await.paginate_backwards(num_events).await
self.inner.write().await?.paginate_backwards(num_events).await
}
/// Paginate forwards in this event-focused timeline, be it room or thread
/// pagination depending on the mode.
pub async fn paginate_forwards(&self, num_events: u16) -> Result<PaginationResult> {
self.inner.write().await.paginate_forwards(num_events).await
self.inner.write().await?.paginate_forwards(num_events).await
}
/// Get the thread root event ID if this linked chunk is in thread mode.
pub async fn thread_root(&self) -> Option<OwnedEventId> {
match &self.inner.read().await.pagination_mode {
pub async fn thread_root(&self) -> Result<Option<OwnedEventId>> {
Ok(match &self.inner.read().await?.pagination_mode {
EventFocusedPaginationMode::Thread { thread_root } => Some(thread_root.clone()),
_ => None,
}
})
}
/// Try to locate the events in the linked chunk corresponding to the given
/// list of decrypted events, and replace them, while alerting observers
/// about the update.
#[cfg(feature = "e2e-encryption")]
pub async fn replace_utds(&self, events: &[ResolvedUtd]) {
let mut guard = self.inner.write().await;
pub async fn replace_utds(&self, events: &[ResolvedUtd]) -> Result<()> {
let mut guard = self.inner.write().await?;
if guard.chunk.replace_utds(events) {
guard.propagate_changes();
guard.notify_subscribers(EventsOrigin::Cache);
}
Ok(())
}
}

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

@@ -265,9 +265,11 @@ impl Caches {
Err(mut event_focused_caches) => {
let cache = event_focused::EventFocusedCache::new(
self.room.weak_room().clone(),
key.focused_event_id.clone(),
key.clone(),
&self.internals.state,
self.internals.linked_chunk_update_sender.clone(),
);
)
.await?;
cache.start_from(number_of_initial_events, thread_mode).await?;
event_focused_caches.insert(key.clone(), cache);

10
crates/matrix-sdk/src/event_cache/redecryptor.rs
View File

@@ -121,11 +121,7 @@ use std::{
use as_variant::as_variant;
use futures_core::Stream;
use futures_util::{
StreamExt,
future::{join_all, try_join_all},
pin_mut,
};
use futures_util::{StreamExt, future::try_join_all, pin_mut};
#[cfg(doc)]
use matrix_sdk_base::{BaseClient, crypto::OlmMachine};
use matrix_sdk_base::{
@@ -460,7 +456,7 @@ impl EventCache {
// event-focused caches alive at the same time, but they could
// accumulate over time. Consider keeping track of which linked chunk
// contains which event ID, to avoid doing the linear searches here.
join_all(
try_join_all(
all_caches
.event_focused
.read()
@@ -468,7 +464,7 @@ impl EventCache {
.values()
.map(|event_focused_cache| event_focused_cache.replace_utds(&events)),
)
.await;
.await?;
}
let report =

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

@@ -31,7 +31,7 @@ use super::{
EventCacheError, EventsOrigin, Result,
caches::{
TimelineVectorDiffs,
event_focused::EventFocusedCacheKey,
event_focused::{EventFocusedCacheKey, EventFocusedCacheState},
pinned_events::PinnedEventsCacheState,
room::{self, RoomEventCacheState},
thread::ThreadEventCacheState,
@@ -40,10 +40,6 @@ use super::{
pub(in super::super) mod selectors;
// Temporary types to make the code compiles. Will be removed one after the
// other.
pub struct EventFocusedCacheState;
/// The type containing all the states, for real.
pub struct State {
store: EventCacheStoreLock,
@@ -331,7 +327,7 @@ impl<'state> ReloadableStateLockWriteGuard<'state> {
async fn reload(&mut self) -> Result<()> {
// Iterate over all states and reload them.
for (room_id, StateForRoom { room, threads, pinned_events, event_focused: _ }) in
for (room_id, StateForRoom { room, threads, pinned_events, event_focused }) in
self.state.by_room.iter_mut()
{
// Room.
@@ -381,6 +377,20 @@ impl<'state> ReloadableStateLockWriteGuard<'state> {
origin: EventsOrigin::Cache,
});
}
// Event-focused.
for event_focused_state in event_focused.values_mut() {
let mut event_focused_state = StateLockWriteGuard {
state: StateLockWriteGuardKind::Reference(event_focused_state),
store: self.store.clone(),
};
let updates_as_vector_diffs = event_focused_state.reload().await?;
let _ = event_focused_state.update_sender.send(TimelineVectorDiffs {
diffs: updates_as_vector_diffs,
origin: EventsOrigin::Cache,
});
}
}
Ok(())

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

@@ -171,6 +171,12 @@ impl From<&PinnedEventsStateSelector> for EventCacheError {
/// Select a [`EventFocusedCacheState`] in [`State`].
pub struct EventFocusedStateSelector(OwnedRoomId, EventFocusedCacheKey);
impl EventFocusedStateSelector {
pub fn new(room_id: OwnedRoomId, key: EventFocusedCacheKey) -> Self {
Self(room_id, key)
}
}
impl CacheState for EventFocusedStateSelector {
type Item = EventFocusedCacheState;