refactor(sdk) Introduce RoomEventCacheStateLock and read/write guards.

This patch extracts fields from `RoomEventCacheState` and move them
into `RoomEventCacheStateLock`. This lock provides 2 methods: `read`
and `write`, respectively to acquire a read-only lock, and a write-only
lock, represented by the `RoomEventCacheStateLockReadGuard` and the
`RoomEventCacheStateLockWriteGuard` types.

All “public” methods on `RoomEventCacheState` now are facade to the read
and write guards.

This refactoring makes the code to compile with the last change in
`EventCacheStore::lock`, which now returns a `EventCacheStoreLockState`.
The next step is to re-load `RoomEventCacheStateLock` when the lock is
dirty! But before doing that, we need this new mechanism to centralise
the management of the store lock.

Author: Hywan

benchmarks/benches/event_cache.rs

@@ -317,6 +317,7 @@ fn find_event_relations(c: &mut Criterion) {
                                 let (target, relations) = room_event_cache
                                     .find_event_with_relations(target_event_id, filter)
                                     .await
+                                    .unwrap()
                                     .unwrap();
                                 assert_eq!(target.event_id().as_deref().unwrap(), target_event_id);
                                 assert_eq!(relations.len(), num_related_events as usize);

benchmarks/benches/room_bench.rs

@@ -181,6 +181,8 @@ pub fn load_pinned_events_benchmark(c: &mut Criterion) {
                 .lock()
                 .await
                 .unwrap()
+                .as_clean()
+                .unwrap()
                 .clear_all_linked_chunks()
                 .await
                 .unwrap();

crates/matrix-sdk-ui/src/timeline/builder.rs

@@ -164,7 +164,7 @@ impl TimelineBuilder {
         room.client().event_cache().subscribe()?;
 
         let (room_event_cache, event_cache_drop) = room.event_cache().await?;
-        let (_, event_subscriber) = room_event_cache.subscribe().await;
+        let (_, event_subscriber) = room_event_cache.subscribe().await.unwrap();
 
         let is_room_encrypted = room
             .latest_encryption_state()
@@ -209,36 +209,36 @@ impl TimelineBuilder {
             .instrument(span)
         });
 
-        let thread_update_join_handle = if let TimelineFocus::Thread { root_event_id: root } =
-            &focus
-        {
-            Some({
-                let span = info_span!(
-                    parent: Span::none(),
-                    "thread_live_update_handler",
-                    room_id = ?room.room_id(),
-                    focus = focus.debug_string(),
-                    prefix = internal_id_prefix
-                );
-                span.follows_from(Span::current());
-
-                // Note: must be done here *before* spawning the task, to avoid race conditions
-                // with event cache updates happening in the background.
-                let (_events, receiver) = room_event_cache.subscribe_to_thread(root.clone()).await;
-
-                spawn(
-                    thread_updates_task(
-                        receiver,
-                        room_event_cache.clone(),
-                        controller.clone(),
-                        root.clone(),
+        let thread_update_join_handle =
+            if let TimelineFocus::Thread { root_event_id: root } = &focus {
+                Some({
+                    let span = info_span!(
+                        parent: Span::none(),
+                        "thread_live_update_handler",
+                        room_id = ?room.room_id(),
+                        focus = focus.debug_string(),
+                        prefix = internal_id_prefix
+                    );
+                    span.follows_from(Span::current());
+
+                    // Note: must be done here *before* spawning the task, to avoid race conditions
+                    // with event cache updates happening in the background.
+                    let (_events, receiver) =
+                        room_event_cache.subscribe_to_thread(root.clone()).await.unwrap();
+
+                    spawn(
+                        thread_updates_task(
+                            receiver,
+                            room_event_cache.clone(),
+                            controller.clone(),
+                            root.clone(),
+                        )
+                        .instrument(span),
                     )
-                    .instrument(span),
-                )
-            })
-        } else {
-            None
-        };
+                })
+            } else {
+                None
+            };
 
         let local_echo_listener_handle = {
             let timeline_controller = controller.clone();

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

@@ -446,7 +446,7 @@ impl<P: RoomDataProvider> TimelineController<P> {
         match focus {
             TimelineFocus::Live { .. } => {
                 // Retrieve the cached events, and add them to the timeline.
-                let events = room_event_cache.events().await;
+                let events = room_event_cache.events().await?;
 
                 let has_events = !events.is_empty();
 
@@ -556,7 +556,8 @@ impl<P: RoomDataProvider> TimelineController<P> {
             }
 
             TimelineFocus::Thread { root_event_id, .. } => {
-                let (events, _) = room_event_cache.subscribe_to_thread(root_event_id.clone()).await;
+                let (events, _) =
+                    room_event_cache.subscribe_to_thread(root_event_id.clone()).await?;
                 let has_events = !events.is_empty();
 
                 // For each event, we also need to find the related events, as they don't
@@ -565,7 +566,7 @@ impl<P: RoomDataProvider> TimelineController<P> {
                 let mut related_events = Vector::new();
                 for event_id in events.iter().filter_map(|event| event.event_id()) {
                     if let Some((_original, related)) =
-                        room_event_cache.find_event_with_relations(&event_id, None).await
+                        room_event_cache.find_event_with_relations(&event_id, None).await?
                     {
                         related_events.extend(related);
                     }

crates/matrix-sdk-ui/src/timeline/pinned_events_loader.rs

@@ -173,7 +173,7 @@ impl PinnedEventsRoom for Room {
         Box::pin(async move {
             if let Ok((cache, _handles)) = self.event_cache().await
                 && let Some(ret) =
-                    cache.find_event_with_relations(event_id, related_event_filters).await
+                    cache.find_event_with_relations(event_id, related_event_filters).await?
             {
                 debug!("Loaded pinned event {event_id} and related events from cache");
                 return Ok(ret);

crates/matrix-sdk-ui/src/timeline/tasks.rs

@@ -28,7 +28,7 @@ use matrix_sdk::{
 use ruma::OwnedEventId;
 use tokio::sync::broadcast::{Receiver, error::RecvError};
 use tokio_stream::StreamExt as _;
-use tracing::{instrument, trace, warn};
+use tracing::{error, instrument, trace, warn};
 
 use crate::timeline::{TimelineController, TimelineFocus, event_item::RemoteEventOrigin};
 
@@ -93,7 +93,14 @@ pub(in crate::timeline) async fn thread_updates_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, _) = room_event_cache.subscribe_to_thread(root.clone()).await;
+                let (initial_events, _) =
+                    match room_event_cache.subscribe_to_thread(root.clone()).await {
+                        Ok(values) => values,
+                        Err(err) => {
+                            error!(?err, "Subscribing to thread failed");
+                            break;
+                        }
+                    };
 
                 timeline_controller
                     .replace_with_initial_remote_events(initial_events, RemoteEventOrigin::Cache)
@@ -145,7 +152,16 @@ pub(in crate::timeline) async fn room_event_cache_updates_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 = room_event_cache.events().await;
+                let initial_events = match room_event_cache.events().await {
+                    Ok(initial_events) => initial_events,
+                    Err(err) => {
+                        error!(
+                            ?err,
+                            "Failed to replace the initial remote events in the event cache"
+                        );
+                        break;
+                    }
+                };
 
                 timeline_controller
                     .replace_with_initial_remote_events(initial_events, RemoteEventOrigin::Cache)

crates/matrix-sdk-ui/tests/integration/timeline/decryption.rs

@@ -73,6 +73,8 @@ async fn test_an_utd_from_the_event_cache_as_an_initial_item_is_decrypted() {
         // The item is an encrypted event! It has been stored before having a chance to
         // be decrypted. Damn. We want to see if decryption will trigger automatically.
         event_cache_store
+            .as_clean()
+            .unwrap()
             .handle_linked_chunk_updates(
                 LinkedChunkId::Room(room_id),
                 vec![
@@ -212,6 +214,8 @@ async fn test_an_utd_from_the_event_cache_as_a_paginated_item_is_decrypted() {
         // chance to be decrypted. Damn. We want to see if decryption will trigger
         // automatically.
         event_cache_store
+            .as_clean()
+            .unwrap()
             .handle_linked_chunk_updates(
                 LinkedChunkId::Room(room_id),
                 vec![

crates/matrix-sdk-ui/tests/integration/timeline/mod.rs

@@ -827,6 +827,8 @@ async fn test_timeline_receives_a_limited_number_of_events_when_subscribing() {
 
         // The event cache contains 30 events.
         event_cache_store
+            .as_clean()
+            .unwrap()
             .handle_linked_chunk_updates(
                 LinkedChunkId::Room(room_id),
                 vec![

crates/matrix-sdk-ui/tests/integration/timeline/pinned_event.rs

@@ -256,7 +256,7 @@ async fn test_cached_events_are_kept_for_different_room_instances() {
     assert!(!items.is_empty()); // We just loaded some events
     assert_pending!(timeline_stream);
 
-    assert!(room_cache.find_event(event_id!("$1")).await.is_some());
+    assert!(room_cache.find_event(event_id!("$1")).await.unwrap().is_some());
 
     // Drop the existing room and timeline instances
     drop(timeline_stream);
@@ -277,7 +277,7 @@ async fn test_cached_events_are_kept_for_different_room_instances() {
 
     let (items, _) = timeline.subscribe().await;
     assert!(!items.is_empty()); // These events came from the cache
-    assert!(room_cache.find_event(event_id!("$1")).await.is_some());
+    assert!(room_cache.find_event(event_id!("$1")).await.unwrap().is_some());
 
     // Drop the existing room and timeline instances
     server.server().reset().await;
@@ -402,7 +402,7 @@ async fn test_pinned_timeline_with_no_pinned_events_on_pagination_is_just_empty(
         .expect("Pagination of events should successful");
 
     // Assert the event is loaded and added to the cache
-    assert!(event_cache.find_event(event_id).await.is_some());
+    assert!(event_cache.find_event(event_id).await.unwrap().is_some());
 
     // And it won't cause an update in the pinned events timeline since it's not
     // pinned

crates/matrix-sdk/src/event_cache/deduplicator.rs

@@ -18,7 +18,7 @@
 use std::collections::BTreeSet;
 
 use matrix_sdk_base::{
-    event_cache::store::EventCacheStoreLock,
+    event_cache::store::EventCacheStoreLockGuard,
     linked_chunk::{LinkedChunkId, Position},
 };
 use ruma::OwnedEventId;
@@ -32,7 +32,7 @@ use super::{
 /// information about the duplicates found in the new events, including the
 /// events that are not loaded in memory.
 pub async fn filter_duplicate_events(
-    store: &EventCacheStoreLock,
+    store_guard: &EventCacheStoreLockGuard,
     linked_chunk_id: LinkedChunkId<'_>,
     linked_chunk: &EventLinkedChunk,
     mut new_events: Vec<Event>,
@@ -50,10 +50,8 @@ pub async fn filter_duplicate_events(
         });
     }
 
-    let store = store.lock().await?;
-
     // Let the store do its magic ✨
-    let duplicated_event_ids = store
+    let duplicated_event_ids = store_guard
         .filter_duplicated_events(
             linked_chunk_id,
             new_events.iter().filter_map(|event| event.event_id()).collect(),
@@ -148,7 +146,10 @@ pub(super) struct DeduplicationOutcome {
 mod tests {
     use std::ops::Not as _;
 
-    use matrix_sdk_base::{deserialized_responses::TimelineEvent, linked_chunk::ChunkIdentifier};
+    use matrix_sdk_base::{
+        deserialized_responses::TimelineEvent, event_cache::store::EventCacheStoreLock,
+        linked_chunk::ChunkIdentifier,
+    };
     use matrix_sdk_test::{async_test, event_factory::EventFactory};
     use ruma::{EventId, owned_event_id, serde::Raw, user_id};
 
@@ -222,6 +223,8 @@ mod tests {
             .unwrap();
 
         let event_cache_store = EventCacheStoreLock::new(event_cache_store, "hodor".to_owned());
+        let event_cache_store = event_cache_store.lock().await.unwrap();
+        let event_cache_store_guard = event_cache_store.as_clean().unwrap();
 
         {
             // When presenting with only duplicate events, some of them in the in-memory
@@ -232,7 +235,7 @@ mod tests {
             linked_chunk.push_events([event_1.clone(), event_2.clone(), event_3.clone()]);
 
             let outcome = filter_duplicate_events(
-                &event_cache_store,
+                event_cache_store_guard,
                 LinkedChunkId::Room(room_id),
                 &linked_chunk,
                 vec![event_0.clone(), event_1.clone(), event_2.clone(), event_3.clone()],
@@ -247,7 +250,7 @@ mod tests {
         linked_chunk.push_events([event_2.clone(), event_3.clone()]);
 
         let outcome = filter_duplicate_events(
-            &event_cache_store,
+            event_cache_store_guard,
             LinkedChunkId::Room(room_id),
             &linked_chunk,
             vec![event_0, event_1, event_2, event_3, event_4],
@@ -351,6 +354,8 @@ mod tests {
 
         // Wrap the store into its lock.
         let event_cache_store = EventCacheStoreLock::new(event_cache_store, "hodor".to_owned());
+        let event_cache_store = event_cache_store.lock().await.unwrap();
+        let event_cache_store_guard = event_cache_store.as_clean().unwrap();
 
         let linked_chunk = EventLinkedChunk::new();
 
@@ -360,7 +365,7 @@ mod tests {
             in_store_duplicated_event_ids,
             non_empty_all_duplicates,
         } = filter_duplicate_events(
-            &event_cache_store,
+            event_cache_store_guard,
             LinkedChunkId::Room(room_id),
             &linked_chunk,
             vec![ev1, ev2, ev3, ev4],