diff --git a/crates/matrix-sdk-ui/src/timeline/controller/mod.rs b/crates/matrix-sdk-ui/src/timeline/controller/mod.rs index f8508f70b..e5eb6a4fc 100644 --- a/crates/matrix-sdk-ui/src/timeline/controller/mod.rs +++ b/crates/matrix-sdk-ui/src/timeline/controller/mod.rs @@ -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); } diff --git a/crates/matrix-sdk-ui/src/timeline/tasks.rs b/crates/matrix-sdk-ui/src/timeline/tasks.rs index 839c862c3..544f8ffee 100644 --- a/crates/matrix-sdk-ui/src/timeline/tasks.rs +++ b/crates/matrix-sdk-ui/src/timeline/tasks.rs @@ -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) diff --git a/crates/matrix-sdk/src/event_cache/caches/event_focused/mod.rs b/crates/matrix-sdk/src/event_cache/caches/event_focused/mod.rs index 54f85f686..3d8342160 100644 --- a/crates/matrix-sdk/src/event_cache/caches/event_focused/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/event_focused/mod.rs @@ -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, @@ -128,7 +137,36 @@ impl EventFocusedCacheState { num_context_events: u16, thread_mode: EventFocusThreadMode, ) -> Result { + 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>> { + 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 { 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, next_gap_token: Option, ) { + // 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>, + inner: Arc>, } 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, - ) -> 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 { + 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> { + pub(super) fn state(&self) -> &CacheStateLock { &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 { + Ok(self.inner.read().await?.update_sender.clone()) } /// Subscribe to updates from this event-focused timeline. - pub async fn subscribe(&self) -> (Vec, Receiver) { - let inner = self.inner.read().await; + pub async fn subscribe(&self) -> Result<(Vec, Receiver)> { + 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 { + 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 { + 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 { - 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 { - 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 { - 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 { - match &self.inner.read().await.pagination_mode { + pub async fn thread_root(&self) -> Result> { + 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(()) } } diff --git a/crates/matrix-sdk/src/event_cache/caches/mod.rs b/crates/matrix-sdk/src/event_cache/caches/mod.rs index a44e415af..d1c3e53fb 100644 --- a/crates/matrix-sdk/src/event_cache/caches/mod.rs +++ b/crates/matrix-sdk/src/event_cache/caches/mod.rs @@ -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); diff --git a/crates/matrix-sdk/src/event_cache/redecryptor.rs b/crates/matrix-sdk/src/event_cache/redecryptor.rs index 2324bca0b..0f100b6b6 100644 --- a/crates/matrix-sdk/src/event_cache/redecryptor.rs +++ b/crates/matrix-sdk/src/event_cache/redecryptor.rs @@ -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 = diff --git a/crates/matrix-sdk/src/event_cache/states/mod.rs b/crates/matrix-sdk/src/event_cache/states/mod.rs index ec7850bb6..aee5749f5 100644 --- a/crates/matrix-sdk/src/event_cache/states/mod.rs +++ b/crates/matrix-sdk/src/event_cache/states/mod.rs @@ -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(()) diff --git a/crates/matrix-sdk/src/event_cache/states/selectors.rs b/crates/matrix-sdk/src/event_cache/states/selectors.rs index b8ae6d68f..f3a1ce5cc 100644 --- a/crates/matrix-sdk/src/event_cache/states/selectors.rs +++ b/crates/matrix-sdk/src/event_cache/states/selectors.rs @@ -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;