matrix-rust-sdk

public

Star0

Fork 0

Watch0

27 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `9ef2e9aaf533`

tarball

atagen <boss@atagen.co>2026-10-25 10:44 UTC

3 files changed+174-29

event-cache: commit history against the window memory holds

commit_history_candidate only diffed incrementally when memory held every stored event, so after a restart, lazy load or shrink each fill and recent load replaced memory with the full candidate and broadcast a Reset of the whole room. It now reloads only the loaded window, extended by chunks the operation created next to it, and sends inserts against what subscribers already hold, falling back to a Reset only when that no longer extends. load_retained_history keeps loading everything through HistoryCommitScope::All.

Modified file: crates/matrix-sdk/src/event_cache/caches/room/history.rs+5 additions−9 deletions@@ -11,7 +11,7 @@ use matrix_sdk_base::{
 use ruma::{OwnedEventId, OwnedTransactionId, TransactionId};
 use tokio::sync::{mpsc, oneshot, watch};
 
-use super::{RoomEventCache, RoomEventCacheInner};
+use super::{HistoryCommitScope, RoomEventCache, RoomEventCacheInner};
 use crate::{
     event_cache::{EventCacheError, Result, caches::event_linked_chunk::EventLinkedChunk},
     room::MessagesOptions,
@@ -277,7 +277,6 @@ impl RoomEventCache {
         {
             let mut state = self.inner.state.write().await?;
             let mut candidate = state.history_candidate().await?;
-            let baseline_count = candidate.events().count();
             let mut known = HashSet::new();
             let mut arrivals = Vec::new();
             let mut leading = Vec::new();
@@ -378,7 +377,7 @@ impl RoomEventCache {
             };
 
             inserted.reverse();
-            state.commit_history_candidate(candidate, inserted, baseline_count).await?;
+            state.commit_history_candidate(candidate, inserted, HistoryCommitScope::Loaded).await?;
             Ok(RecentHistoryOutcome { pages, events: raw_count, continuity })
         }
     }
@@ -397,8 +396,7 @@ impl RoomEventCache {
         if !gaps(&candidate).is_empty()
             && candidate.chunks().count() != state.room_linked_chunk().chunks().count()
         {
-            let baseline_count = candidate.events().count();
-            state.commit_history_candidate(candidate, Vec::new(), baseline_count).await?;
+            state.commit_history_candidate(candidate, Vec::new(), HistoryCommitScope::All).await?;
         }
         Ok(())
     }
@@ -436,7 +434,6 @@ impl RoomEventCache {
         {
             let mut state = self.inner.state.write().await?;
             let mut candidate = state.history_candidate().await?;
-            let baseline_count = candidate.events().count();
             let older_boundary = candidate.revents().find_map(|(_, event)| event.event_id());
             let mut known: HashSet<_> =
                 candidate.events().filter_map(|(_, event)| event.event_id()).collect();
@@ -464,7 +461,7 @@ impl RoomEventCache {
                 None
             };
             candidate.push_live_events(gap, &events);
-            state.commit_history_candidate(candidate, events, baseline_count).await?;
+            state.commit_history_candidate(candidate, events, HistoryCommitScope::Loaded).await?;
             Ok(history)
         }
     }
@@ -513,7 +510,6 @@ impl RoomEventCache {
         {
             let mut state = self.inner.state.write().await?;
             let mut candidate = state.history_candidate().await?;
-            let baseline_count = candidate.events().count();
             let Some(current) = find_gap(&candidate, &gap_id) else {
                 return Ok(HistoryGapFillOutcome::NotFound);
             };
@@ -537,7 +533,7 @@ impl RoomEventCache {
                 failure.as_ref().map(ToString::to_string),
             );
             let count = events.len();
-            state.commit_history_candidate(candidate, events, baseline_count).await?;
+            state.commit_history_candidate(candidate, events, HistoryCommitScope::Loaded).await?;
             if let Some(error) = failure {
                 return Err(EventCacheError::PaginationError(std::sync::Arc::new(error)));
             }

Modified file: crates/matrix-sdk/src/event_cache/caches/room/mod.rs+71 additions@@ -41,6 +41,7 @@ use ruma::{
     events::{AnyRoomAccountDataEvent, AnySyncEphemeralRoomEvent, relation::RelationType},
     serde::Raw,
 };
+use state::HistoryCommitScope;
 pub(super) use state::{LockedRoomEventCacheState, RoomEventCacheStateLockWriteGuard};
 pub use subscriber::RoomEventCacheSubscriber;
 use tokio::sync::{Notify, broadcast::Receiver, mpsc};
@@ -1970,6 +1971,76 @@ mod timed_tests {
     }
 
     #[async_test]
+    async fn test_history_commit_after_shrink_stays_within_the_loaded_window() {
+        let room_id = room_id!("!shrunk-window:saucisse.bzh");
+        let client = MockClientBuilder::new(None).build().await;
+        let event_cache = client.event_cache();
+        event_cache.subscribe().unwrap();
+        client.base_client().get_or_create_room(room_id, RoomState::Joined);
+        let room = client.get_room(room_id).unwrap();
+        let (room_event_cache, _drop_handles) = room.event_cache().await.unwrap();
+        let f = EventFactory::new().room(room_id).sender(*ALICE);
+        let old_id = event_id!("$window-old");
+        let new_id = event_id!("$window-new");
+        let newer_id = event_id!("$window-newer");
+
+        room_event_cache
+            .handle_joined_room_update(JoinedRoomUpdate {
+                timeline: Timeline {
+                    events: vec![f.text_msg("old").event_id(old_id).into_event()],
+                    ..Default::default()
+                },
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        room_event_cache
+            .handle_joined_room_update(JoinedRoomUpdate {
+                timeline: Timeline {
+                    limited: true,
+                    prev_batch: Some("new".to_owned()),
+                    events: vec![f.text_msg("new").event_id(new_id).into_event()],
+                },
+                ..Default::default()
+            })
+            .await
+            .unwrap();
+        {
+            let mut state = room_event_cache.inner.state.write().await.unwrap();
+            state.shrink_to_last_chunk().await.unwrap();
+            let _ = state.room_linked_chunk_mut().updates_as_vector_diffs();
+        }
+
+        let (events, mut subscriber) = room_event_cache.subscribe().await.unwrap();
+        let ids: Vec<_> = events.iter().filter_map(|event| event.event_id()).collect();
+        assert_eq!(ids, [new_id]);
+
+        room_event_cache
+            .append_history_section(
+                vec![f.text_msg("newer").event_id(newer_id).into_event()],
+                Some("newer".to_owned()),
+            )
+            .await
+            .unwrap();
+
+        assert_let_timeout!(
+            Ok(RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs, .. })) =
+                subscriber.recv()
+        );
+        assert_matches!(diffs.as_slice(), [VectorDiff::Insert { index: 1, value }] => {
+            assert_eq!(value.event_id().as_deref(), Some(newer_id));
+        });
+        let ids: Vec<_> = room_event_cache
+            .events()
+            .await
+            .unwrap()
+            .iter()
+            .filter_map(|event| event.event_id())
+            .collect();
+        assert_eq!(ids, [new_id, newer_id]);
+    }
+
+    #[async_test]
     async fn test_limited_sync_after_shrink_to_a_gap_keeps_its_discontinuity() {
         let room_id = room_id!("!shrunk-history:saucisse.bzh");
         let client = MockClientBuilder::new(None).build().await;

Modified file: crates/matrix-sdk/src/event_cache/caches/room/state.rs+98 additions−20 deletions@@ -30,8 +30,8 @@ use matrix_sdk_base::{
         store::{EventCacheStoreLock, EventCacheStoreLockGuard, EventCacheStoreLockState},
     },
     linked_chunk::{
-        ChunkContent, ChunkMetadata, LinkedChunkId, OwnedLinkedChunkId, Position, Update,
-        lazy_loader,
+        ChunkContent, ChunkIdentifier, ChunkMetadata, LinkedChunkId, OwnedLinkedChunkId, Position,
+        Update, lazy_loader,
     },
     serde_helpers::{extract_edit_target, extract_thread_root},
     sync::Timeline,
@@ -775,25 +775,36 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
     }
 
     pub(super) async fn history_candidate(&self) -> Result<EventLinkedChunk, EventCacheError> {
+        self.load_from_store(None).await
+    }
+
+    /// Loads stored chunks from the newest back to `first`, or all of them.
+    async fn load_from_store(
+        &self,
+        first: Option<ChunkIdentifier>,
+    ) -> Result<EventLinkedChunk, EventCacheError> {
         let linked_chunk_id = LinkedChunkId::Room(&self.state.room_id);
         let (last, generator) = self.store.load_last_chunk(linked_chunk_id).await?;
         let metadata = self.store.load_all_chunks_metadata(linked_chunk_id).await?;
-        let mut candidate = EventLinkedChunk::with_initial_linked_chunk(
+        let mut loaded = EventLinkedChunk::with_initial_linked_chunk(
             lazy_loader::from_last_chunk(last, generator)?,
             (!metadata.is_empty()).then_some(metadata),
         );
         loop {
-            let first =
-                candidate.chunks().next().expect("a linked chunk is never empty").identifier();
-            let Some(previous) = self.store.load_previous_chunk(linked_chunk_id, first).await?
+            let current =
+                loaded.chunks().next().expect("a linked chunk is never empty").identifier();
+            if Some(current) == first {
+                break;
+            }
+            let Some(previous) = self.store.load_previous_chunk(linked_chunk_id, current).await?
             else {
                 break;
             };
-            candidate.insert_new_chunk_as_first(previous)?;
+            loaded.insert_new_chunk_as_first(previous)?;
         }
-        let _ = candidate.store_updates().take();
-        let _ = candidate.updates_as_vector_diffs();
-        Ok(candidate)
+        let _ = loaded.store_updates().take();
+        let _ = loaded.updates_as_vector_diffs();
+        Ok(loaded)
     }
 
     pub(crate) fn history_gap_positions(&self) -> Vec<super::HistoryGapPosition> {
@@ -825,20 +836,33 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
         &mut self,
         mut candidate: EventLinkedChunk,
         events: Vec<Event>,
-        baseline_count: usize,
+        scope: HistoryCommitScope,
     ) -> Result<(), EventCacheError> {
-        let use_incremental = self.state.room_linked_chunk.events().count() == baseline_count;
-        self.send_updates_to_store(candidate.store_updates().take()).await?;
-        self.state.room_linked_chunk = candidate;
+        let previous: Vec<_> =
+            self.state.room_linked_chunk.events().map(|(_, event)| event.event_id()).collect();
+        let mut diffs = match scope {
+            HistoryCommitScope::Loaded => {
+                let start = self.loaded_window_start(&candidate).await?;
+                self.send_updates_to_store(candidate.store_updates().take()).await?;
+                self.state.room_linked_chunk = self.load_from_store(start).await?;
+                insertions_from(&previous, &self.state.room_linked_chunk)
+            }
+            HistoryCommitScope::All => {
+                self.send_updates_to_store(candidate.store_updates().take()).await?;
+                let _ = candidate.updates_as_vector_diffs();
+                self.state.room_linked_chunk = candidate;
+                None
+            }
+        }
+        .unwrap_or_else(|| {
+            let values =
+                self.state.room_linked_chunk.events().map(|(_, event)| event.clone()).collect();
+            vec![VectorDiff::Reset { values }]
+        });
         self.reset_pagination_status(false);
         let result =
             self.post_process_new_events(events, PostProcessingOrigin::Backpagination, None).await;
-        let mut diffs = self.state.room_linked_chunk.updates_as_vector_diffs();
-        if !use_incremental {
-            let values =
-                self.state.room_linked_chunk.events().map(|(_, event)| event.clone()).collect();
-            diffs = vec![VectorDiff::Reset { values }];
-        }
+        diffs.extend(self.state.room_linked_chunk.updates_as_vector_diffs());
         self.state.update_sender.send(
             RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
                 diffs,
@@ -850,6 +874,34 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
         result
     }
 
+    /// The first chunk of `candidate` that belongs to the window memory already
+    /// loaded. A chunk created by the operation joins the window of the stored
+    /// chunk after it, so a page filled into a gap right before the window shows,
+    /// while history further back stays unloaded.
+    async fn loaded_window_start(
+        &self,
+        candidate: &EventLinkedChunk,
+    ) -> Result<Option<ChunkIdentifier>, EventCacheError> {
+        let stored: HashSet<_> = self
+            .store
+            .load_all_chunks_metadata(LinkedChunkId::Room(&self.state.room_id))
+            .await?
+            .into_iter()
+            .map(|metadata| metadata.identifier)
+            .collect();
+        let loaded: HashSet<_> =
+            self.state.room_linked_chunk.chunks().map(|chunk| chunk.identifier()).collect();
+        let mut start = None;
+        for chunk in candidate.rchunks() {
+            let identifier = chunk.identifier();
+            if stored.contains(&identifier) && !loaded.contains(&identifier) {
+                break;
+            }
+            start = Some(identifier);
+        }
+        Ok(start)
+    }
+
     fn reset_pagination_status(&self, hit_timeline_start: bool) {
         let mut status = self.state.pagination_status.write();
         self.state.generation.fetch_add(1, Ordering::SeqCst);
@@ -1523,6 +1575,32 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
     }
 }
 
+/// How much of the committed history ends up in memory.
+pub(super) enum HistoryCommitScope {
+    /// Keep the window memory already held, plus pages added right next to it.
+    Loaded,
+    /// Load every stored section, as before attaching a fresh subscriber.
+    All,
+}
+
+/// Insert diffs turning `previous` into the events now loaded, or `None` when
+/// the loaded events no longer extend it, which calls for a reset.
+fn insertions_from(
+    previous: &[Option<OwnedEventId>],
+    loaded: &EventLinkedChunk,
+) -> Option<Vec<VectorDiff<Event>>> {
+    let mut remaining = previous.iter().peekable();
+    let mut diffs = Vec::new();
+    for (index, (_, event)) in loaded.events().enumerate() {
+        if remaining.peek().is_some_and(|id| **id == event.event_id()) {
+            remaining.next();
+        } else {
+            diffs.push(VectorDiff::Insert { index, value: event.clone() });
+        }
+    }
+    remaining.next().is_none().then_some(diffs)
+}
+
 /// Extract a valid read receipt event from the ephemeral events, if
 /// available.
 fn extract_read_receipt(

matrix-rust-sdk

Commit 9ef2e9aaf533

Commit `9ef2e9aaf533`