fix(MemoryStore): undeadlock timeline saving

Merge pull request #509 from matrix-org/gnunicorn/issue508

Author: gnunicorn

crates/matrix-sdk-base/Cargo.toml

@@ -36,7 +36,7 @@ indexeddb_cryptostore = ["matrix-sdk-crypto/indexeddb_cryptostore"]
 [dependencies]
 async-stream = "0.3.2"
 chacha20poly1305 = { version = "0.9.0", optional = true }
-dashmap = "4.0.2"
+dashmap = "5.1.0"
 futures-core = "0.3.15"
 futures-util = { version = "0.3.15", default-features = false }
 futures-channel = "0.3.15"

crates/matrix-sdk-base/src/rooms/normal.rs

@@ -497,7 +497,11 @@ impl Room {
         {
             TimelineStreamBackward::new(event_ids.clone(), end_token, Some(stored_events))
         } else {
-            TimelineStreamBackward::new(event_ids.clone(), Some(sync_token.clone().unwrap()), None)
+            TimelineStreamBackward::new(
+                event_ids.clone(),
+                Some(sync_token.clone().expect("Sync token exists")),
+                None,
+            )
         };
 
         backward_timeline_streams.push(backward_sender);

crates/matrix-sdk-base/src/store/memory_store.rs

@@ -284,10 +284,10 @@ impl MemoryStore {
                 info!("Save new timeline batch from messages response for {}", room);
             }
 
-            let data = if timeline.limited {
+            let mut delete_timeline = false;
+            if timeline.limited {
                 info!("Delete stored timeline for {} because the sync response was limited", room);
-                self.room_timeline.remove(room);
-                None
+                delete_timeline = true;
             } else if let Some(mut data) = self.room_timeline.get_mut(room) {
                 if !timeline.sync && Some(&timeline.start) != data.end.as_ref() {
                     // This should only happen when a developer adds a wrong timeline
@@ -298,7 +298,6 @@ impl MemoryStore {
                 }
 
                 // Check if the event already exists in the store
-                let mut delete_timeline = false;
                 for event in &timeline.events {
                     if let Some(event_id) = event.event_id() {
                         if data.event_id_to_position.contains_key(&event_id) {
@@ -308,37 +307,39 @@ impl MemoryStore {
                     }
                 }
 
-                if delete_timeline {
-                    info!("Delete stored timeline for {} because of duplicated events", room);
-                    self.room_timeline.remove(room);
-                    None
-                } else if timeline.sync {
-                    data.start = timeline.start.clone();
-                    Some(data)
-                } else {
-                    data.end = timeline.end.clone();
-                    Some(data)
+                if !delete_timeline {
+                    if timeline.sync {
+                        data.start = timeline.start.clone();
+                    } else {
+                        data.end = timeline.end.clone();
+                    }
                 }
-            } else {
-                None
-            };
+            }
 
-            let mut data = &mut *if let Some(data) = data {
-                data
-            } else {
-                let data = TimelineData {
+            if delete_timeline {
+                info!("Delete stored timeline for {} because of duplicated events", room);
+                self.room_timeline.remove(room);
+            }
+
+            let mut data =
+                self.room_timeline.entry(room.to_owned()).or_insert_with(|| TimelineData {
                     start: timeline.start.clone(),
                     end: timeline.end.clone(),
                     ..Default::default()
-                };
-                self.room_timeline.insert(room.to_owned(), data);
-                self.room_timeline.get_mut(room).unwrap()
+                });
+
+            let make_room_version = || {
+                self.room_info
+                    .get(room)
+                    .and_then(|info| {
+                        info.base_info.create.as_ref().map(|event| event.room_version.clone())
+                    })
+                    .unwrap_or_else(|| {
+                        warn!("Unable to find the room version for {}, assume version 9", room);
+                        RoomVersionId::V9
+                    })
             };
 
-            // Create a copy of the events if the stream created via `room_timeline()` isn't
-            // fully consumed
-            let data_events = Arc::make_mut(&mut data.events);
-
             if timeline.sync {
                 let mut room_version = None;
                 for event in timeline.events.iter().rev() {
@@ -347,21 +348,13 @@ impl MemoryStore {
                         redaction,
                     ))) = event.event.deserialize()
                     {
-                        if let Some(position) = data.event_id_to_position.get(&redaction.redacts) {
-                            if let Some(mut full_event) = data_events.get_mut(position) {
+                        let pos = data.event_id_to_position.get(&redaction.redacts).copied();
+
+                        if let Some(position) = pos {
+                            if let Some(mut full_event) = data.events.get_mut(&position.clone()) {
                                 let inner_event = full_event.event.deserialize()?;
                                 if room_version.is_none() {
-                                    room_version = Some(self.room_info
-                                                        .get(room)
-                                                        .and_then(|info| {
-                                                            info.base_info
-                                                                .create
-                                                                .as_ref()
-                                                                .map(|event| event.room_version.clone())
-                                                        }).unwrap_or_else(|| {
-                                                            warn!("Unable to find the room version for {}, assume version 9", room);
-                                                            RoomVersionId::V9
-                                                        }));
+                                    room_version = Some(make_room_version());
                                 }
 
                                 full_event.event = Raw::new(&AnySyncRoomEvent::from(
@@ -372,20 +365,22 @@ impl MemoryStore {
                     }
 
                     data.start_position -= 1;
+                    let start_position = data.start_position;
                     // Only add event with id to the position map
                     if let Some(event_id) = event.event_id() {
-                        data.event_id_to_position.insert(event_id, data.start_position);
+                        data.event_id_to_position.insert(event_id, start_position);
                     }
-                    data_events.insert(data.start_position, event.to_owned());
+                    data.events.insert(start_position, event.clone());
                 }
             } else {
                 for event in timeline.events.iter() {
                     data.end_position += 1;
+                    let end_position = data.end_position;
                     // Only add event with id to the position map
                     if let Some(event_id) = event.event_id() {
-                        data.event_id_to_position.insert(event_id, data.end_position);
+                        data.event_id_to_position.insert(event_id, end_position);
                     }
-                    data_events.insert(data.end_position, event.to_owned());
+                    data.events.insert(end_position, event.clone());
                 }
             }
         }
@@ -580,22 +575,22 @@ impl MemoryStore {
         &self,
         room_id: &RoomId,
     ) -> Result<Option<(BoxStream<Result<SyncRoomEvent>>, Option<String>)>> {
-        if let Some(data) = self.room_timeline.get(room_id) {
-            let events = data.events.clone();
-            let stream = stream! {
-                for item in events.values() {
-                    yield Ok(item.to_owned());
-                }
-            };
-            info!(
-                "Found previously stored timeline for {}, with end token {:?}",
-                room_id, data.end
-            );
-            Ok(Some((Box::pin(stream), data.end.to_owned())))
+        let (events, end_token) = if let Some(data) = self.room_timeline.get(room_id) {
+            (data.events.clone(), data.end.clone())
         } else {
             info!("No timeline for {} was previously stored", room_id);
-            Ok(None)
-        }
+            return Ok(None);
+        };
+
+        let stream = stream! {
+            for (_, item) in events {
+                yield Ok(item);
+            }
+        };
+
+        info!("Found previously stored timeline for {}, with end token {:?}", room_id, end_token);
+
+        Ok(Some((Box::pin(stream), end_token)))
     }
 }
 
@@ -762,7 +757,7 @@ struct TimelineData {
     pub start_position: isize,
     pub end: Option<String>,
     pub end_position: isize,
-    pub events: Arc<BTreeMap<isize, SyncRoomEvent>>,
+    pub events: BTreeMap<isize, SyncRoomEvent>,
     pub event_id_to_position: HashMap<Box<EventId>, isize>,
 }
 

crates/matrix-sdk-base/src/store/sled_store.rs

@@ -20,6 +20,7 @@ use std::{
     time::Instant,
 };
 
+use async_stream::stream;
 use futures_core::stream::Stream;
 use futures_util::stream::{self, TryStreamExt};
 use matrix_sdk_common::async_trait;
@@ -1012,25 +1013,33 @@ impl SledStore {
     ) -> Result<Option<(BoxStream<Result<SyncRoomEvent>>, Option<String>)>> {
         let db = self.clone();
         let key = room_id.encode();
+        let r_id = room_id.to_owned();
         let metadata: Option<TimelineMetadata> = db
             .room_timeline_metadata
             .get(key.as_slice())?
             .map(|v| serde_json::from_slice(&v).map_err(StoreError::Json))
             .transpose()?;
-        if metadata.is_none() {
-            info!("No timeline for {} was previously stored", room_id);
-            return Ok(None);
-        }
-        let end_token = metadata.and_then(|m| m.end);
-        let stream = Box::pin(stream::iter(
-            db.room_timeline
-                .scan_prefix(key)
-                .map(move |v| db.deserialize_event(&v?.1).map_err(|e| e.into())),
-        ));
+        let metadata = match metadata {
+            Some(m) => m,
+            None => {
+                info!("No timeline for {} was previously stored", r_id);
+                return Ok(None);
+            }
+        };
 
-        info!("Found previously stored timeline for {}, with end token {:?}", room_id, end_token);
+        let mut position = metadata.start_position;
+        let end_token = metadata.end;
+
+        info!("Found previously stored timeline for {}, with end token {:?}", r_id, end_token);
+
+        let stream = stream! {
+            while let Ok(Some(item)) = db.room_timeline.get(&(r_id.as_ref(), position).encode()) {
+                position += 1;
+                yield db.deserialize_event(&item).map_err(|e| e.into());
+            }
+        };
 
-        Ok(Some((stream, end_token)))
+        Ok(Some((Box::pin(stream), end_token)))
     }
 
     async fn remove_room_timeline(&self, room_id: &RoomId) -> Result<()> {