matrix-rust-sdk

public

Star0

Fork 0

Watch0

49 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `ba8ba5be174f`

tarball

atagen <boss@atagen.co>2026-10-27 05:11 UTC

4 files changed+215-18

event-cache: never trust our own sends to mark where history joins

`insert_sent_event_from_send_queue` appends a sent event at the live end before sync delivers what others sent ahead of it, so its retained position is a guess. As the newest retained event it made a limited sync that contained it drop its `prev_batch`, and gap fills, recent loads and the hole probe joined on it, so the events before it were lost with no gap left. Our events now anchor only in rooms where nobody else's event is retained, and a server page re-places them otherwise.

Modified file: crates/matrix-sdk/src/event_cache/caches/room/history.rs+66 additions−10 deletions@@ -8,7 +8,7 @@ use matrix_sdk_base::{
     event_cache::{Event, Gap},
     linked_chunk::ChunkContent,
 };
-use ruma::{OwnedEventId, OwnedTransactionId, TransactionId};
+use ruma::{OwnedEventId, OwnedTransactionId, TransactionId, UserId};
 use tokio::sync::{mpsc, oneshot, watch};
 
 use super::{HistoryCommitScope, RoomEventCache, RoomEventCacheInner};
@@ -258,16 +258,21 @@ impl RoomEventCache {
         &self,
         mut cancellation: watch::Receiver<bool>,
     ) -> Result<RecentHistoryOutcome> {
-        let (anchor, retained) = {
+        let room = self.inner.weak_room.get().ok_or(EventCacheError::ClientDropped)?;
+        let (anchor, retained, anchors) = {
             let state = self.inner.state.write().await?;
+            let anchors = Anchors::new(
+                room.own_user_id(),
+                state.room_linked_chunk().events().map(|(_, event)| event),
+            );
             let retained: HashSet<_> = state
                 .room_linked_chunk()
                 .events()
+                .filter(|(_, event)| anchors.holds(event))
                 .filter_map(|(_, event)| event.event_id())
                 .collect();
-            (state.newest_retained_event_id().await?, retained)
+            (state.newest_retained_event_id().await?, retained, anchors)
         };
-        let room = self.inner.weak_room.get().ok_or(EventCacheError::ClientDropped)?;
         let mut cursor = None;
         let mut visited = Vec::new();
         let mut events = Vec::new();
@@ -293,7 +298,10 @@ impl RoomEventCache {
             let mut page = response.chunk;
             if let Some(index) = anchor.as_ref().and_then(|anchor| {
                 page.iter().position(|event| {
-                    event.event_id().is_some_and(|id| &id == anchor || retained.contains(&id))
+                    anchors.holds(event)
+                        && event
+                            .event_id()
+                            .is_some_and(|id| &id == anchor || retained.contains(&id))
                 })
             }) {
                 page.truncate(index);
@@ -472,22 +480,27 @@ impl RoomEventCache {
         let mut candidate = state.history_candidate().await?;
         let known: HashSet<_> =
             candidate.events().filter_map(|(_, event)| event.event_id()).collect();
+        let anchors = Anchors::new(room.own_user_id(), candidate.events().map(|(_, event)| event));
         let mut joined = false;
         let mut missing = Vec::new();
         for event in response.chunk {
             match event.event_id() {
-                Some(id) if known.contains(&id) => {
+                Some(id) if known.contains(&id) && anchors.holds(&event) => {
                     joined = true;
                     break;
                 }
                 _ => missing.push(event),
             }
         }
-        let count = missing.len();
+        let count = missing
+            .iter()
+            .filter(|event| event.event_id().is_none_or(|id| !known.contains(&id)))
+            .count();
         tracing::debug!(room_id = %self.room_id(), new = count, joined, "history gap probe");
-        if missing.is_empty() {
+        if count == 0 {
             return Ok(Some(0));
         }
+        anchors.unplace(&mut candidate, &missing);
         missing.reverse();
         let gap = (!joined).then(|| response.end).flatten().map(|token| Gap {
             token,
@@ -676,6 +689,7 @@ impl RoomEventCache {
             };
             let (gap, events) = apply_page(
                 &mut candidate,
+                room.own_user_id(),
                 &gap_id,
                 events,
                 next,
@@ -691,6 +705,40 @@ impl RoomEventCache {
     }
 }
 
+pub(super) fn from_others(own_user_id: &UserId, event: &Event) -> bool {
+    event.sender().is_some_and(|sender| sender != own_user_id)
+}
+
+/// Which retained events show where a server page meets retained history. The send queue
+/// appends our own sends at the live end, ahead of whatever sync has yet to deliver, so ours
+/// only count while nobody else's event is retained, and a page re-places them otherwise.
+#[derive(Clone, Copy)]
+struct Anchors<'a> {
+    own_user_id: &'a UserId,
+    ours: bool,
+}
+
+impl<'a> Anchors<'a> {
+    fn new<'e>(own_user_id: &'a UserId, retained: impl IntoIterator<Item = &'e Event>) -> Self {
+        let ours = !retained.into_iter().any(|event| from_others(own_user_id, event));
+        Self { own_user_id, ours }
+    }
+
+    fn holds(&self, event: &Event) -> bool {
+        self.ours || from_others(self.own_user_id, event)
+    }
+
+    fn unplace(&self, candidate: &mut EventLinkedChunk, page: &[Event]) {
+        let positions = page
+            .iter()
+            .filter(|event| !self.holds(event))
+            .filter_map(|event| candidate.find_event(&event.event_id()?))
+            .map(|(position, _)| position)
+            .collect();
+        candidate.remove_events_by_position(positions).expect("positions were just found");
+    }
+}
+
 fn insert_run(
     candidate: &mut EventLinkedChunk,
     before: Option<&ruma::EventId>,
@@ -743,6 +791,7 @@ fn find_gap(candidate: &EventLinkedChunk, gap_id: &TransactionId) -> Option<Gap>
 
 fn apply_page(
     candidate: &mut EventLinkedChunk,
+    own_user_id: &UserId,
     gap_id: &TransactionId,
     mut events: Vec<Event>,
     next: Option<String>,
@@ -757,14 +806,20 @@ fn apply_page(
     let history = gap.history.as_mut().expect("the gap was checked under the state lock");
     // `/messages` may never return the boundary itself (filtered, redacted, or ordered apart
     // from sync arrival), so any event already below the gap proves the page reached it
-    let below: HashSet<_> = candidate
+    let below_events: Vec<_> = candidate
         .chunks()
         .take_while(|chunk| chunk.identifier() != chunk_id)
         .filter_map(|chunk| match chunk.content() {
-            ChunkContent::Items(items) => Some(items.iter().filter_map(Event::event_id)),
+            ChunkContent::Items(items) => Some(items.iter()),
             ChunkContent::Gap(_) => None,
         })
         .flatten()
+        .collect();
+    let anchors = Anchors::new(own_user_id, below_events.iter().copied());
+    let below: HashSet<_> = below_events
+        .into_iter()
+        .filter(|event| anchors.holds(event))
+        .filter_map(Event::event_id)
         .chain(history.older_boundary.clone())
         .collect();
     let overlap =
@@ -791,6 +846,7 @@ fn apply_page(
     if let Some(next) = next {
         gap.token = next;
     }
+    anchors.unplace(candidate, &events);
     let mut known: HashSet<_> =
         candidate.events().filter_map(|(_, event)| event.event_id()).collect();
     events.retain(|event| event.event_id().is_none_or(|id| known.insert(id)));

Modified file: crates/matrix-sdk/src/event_cache/caches/room/state.rs+26 additions−7 deletions@@ -859,8 +859,25 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
     pub(super) async fn newest_retained_event_id(
         &self,
     ) -> Result<Option<OwnedEventId>, EventCacheError> {
+        self.newest_retained_event_id_where(|_| true).await
+    }
+
+    async fn newest_retained_event_id_from_others(
+        &self,
+    ) -> Result<Option<OwnedEventId>, EventCacheError> {
+        let own_user_id = &self.state.own_user_id;
+        self.newest_retained_event_id_where(|event| super::history::from_others(own_user_id, event))
+            .await
+    }
+
+    async fn newest_retained_event_id_where(
+        &self,
+        keep: impl Fn(&Event) -> bool,
+    ) -> Result<Option<OwnedEventId>, EventCacheError> {
         let chunk = &self.state.room_linked_chunk;
-        if let Some(id) = chunk.revents().find_map(|(_, event)| event.event_id()) {
+        if let Some(id) =
+            chunk.revents().filter(|(_, event)| keep(event)).find_map(|(_, event)| event.event_id())
+        {
             return Ok(Some(id));
         }
         let linked_chunk_id = LinkedChunkId::Room(&self.state.room_id);
@@ -870,7 +887,8 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
         let mut seen = HashSet::from([first]);
         while let Some(previous) = self.store.load_previous_chunk(linked_chunk_id, first).await? {
             if let ChunkContent::Items(events) = &previous.content
-                && let Some(id) = events.iter().rev().find_map(Event::event_id)
+                && let Some(id) =
+                    events.iter().rev().filter(|event| keep(event)).find_map(Event::event_id)
             {
                 return Ok(Some(id));
             }
@@ -878,10 +896,7 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
             if !seen.insert(first) {
                 error!(room_id = %self.state.room_id, "stored linked chunk loops back on itself");
                 return Err(EventCacheError::LinkedChunkLoader(
-                    lazy_loader::LazyLoaderError::Cycle {
-                        new_chunk: first,
-                        with_chunk: first,
-                    },
+                    lazy_loader::LazyLoaderError::Cycle { new_chunk: first, with_chunk: first },
                 ));
             }
         }
@@ -1071,7 +1086,11 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
         }
 
         let retained = newest_retained.is_some();
-        let older_boundary = timeline.limited.then_some(newest_retained).flatten();
+        let older_boundary = if timeline.limited {
+            self.newest_retained_event_id_from_others().await?
+        } else {
+            None
+        };
         let newer_boundary =
             timeline.limited.then(|| events.iter().find_map(Event::event_id)).flatten();
         if older_boundary.is_some() && events.iter().any(|event| event.event_id() == older_boundary)

Modified file: crates/matrix-sdk/tests/recent_history.rs+49 additions@@ -773,3 +773,52 @@ async fn test_a_probe_behind_a_non_limited_sync_brings_back_what_it_skipped() {
         .await;
     assert_eq!(cache.probe_sync_hole().await.unwrap(), Some(0));
 }
+
+#[async_test]
+async fn test_recent_history_does_not_join_on_our_own_send() {
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+    client.event_cache().subscribe().unwrap();
+    let room_id = room_id!("!sent:example.org");
+    let room = server.sync_joined_room(&client, room_id).await;
+    let (cache, _handles) = room.event_cache().await.unwrap();
+    let f = event_factory(room_id);
+    let own_user_id = client.user_id().unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("first").event_id(event_id!("$first"))),
+        )
+        .await;
+    server.mock_room_state_encryption().plain().mount().await;
+    server.mock_room_send().ok(event_id!("$mine")).mock_once().mount().await;
+    room.send_queue()
+        .send(ruma::events::room::message::RoomMessageEventContent::text_plain("mine").into())
+        .await
+        .unwrap();
+    tokio::time::timeout(Duration::from_secs(5), async {
+        while cache.events().await.unwrap().len() < 2 {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+    server
+        .mock_room_messages()
+        .match_limit(50)
+        .ok(RoomMessagesResponseTemplate::default().end_token("unused").events(vec![
+            f.text_msg("mine").sender(own_user_id).event_id(event_id!("$mine")),
+            f.text_msg("theirs").event_id(event_id!("$theirs")),
+            f.text_msg("first").event_id(event_id!("$first")),
+        ]))
+        .mock_once()
+        .mount()
+        .await;
+
+    let outcome = cache.load_recent_history().await.unwrap();
+    assert_eq!(outcome.continuity, RecentHistoryContinuity::Joined);
+    let ids: Vec<_> =
+        cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();
+    assert_eq!(ids, [event_id!("$first"), event_id!("$theirs"), event_id!("$mine")]);
+}

Modified file: crates/matrix-sdk/tests/segmented_history.rs+74 additions−1 deletions@@ -836,8 +836,81 @@ async fn test_cold_open_pagination_keeps_the_order_tracker_in_link_order() {
     cache.pagination().run_backwards_once(20).await.unwrap();
     let ids: Vec<_> =
         cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();
+    assert_eq!(ids, vec![event_id!("$y"), event_id!("$z"), event_id!("$a"), event_id!("$live")]);
+}
+
+#[async_test]
+async fn test_a_gap_after_our_send_fills_what_others_sent_before_it() {
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+    client.event_cache().subscribe().unwrap();
+    let room_id = room_id!("!sent:example.org");
+    let room = server.sync_joined_room(&client, room_id).await;
+    let (cache, _handles) = room.event_cache().await.unwrap();
+    let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
+    let own_user_id = client.user_id().unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("first").event_id(event_id!("$first")))
+                .set_timeline_prev_batch("old-edge"),
+        )
+        .await;
+    server.mock_room_state_encryption().plain().mount().await;
+    server.mock_room_send().ok(event_id!("$mine")).mock_once().mount().await;
+    room.send_queue()
+        .send(ruma::events::room::message::RoomMessageEventContent::text_plain("mine").into())
+        .await
+        .unwrap();
+    tokio::time::timeout(Duration::from_secs(5), async {
+        while !cache
+            .events()
+            .await
+            .unwrap()
+            .iter()
+            .any(|event| event.event_id().as_deref() == Some(event_id!("$mine")))
+        {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap();
+
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("later").event_id(event_id!("$later")))
+                .set_timeline_limited()
+                .set_timeline_prev_batch("before-later"),
+        )
+        .await;
+    let gaps = cache.history_gaps().await.unwrap();
+    assert_eq!(gaps.len(), 1);
+    assert_eq!(gaps[0].older_boundary.as_deref(), Some(event_id!("$first")));
+
+    server
+        .mock_room_messages()
+        .match_from("before-later")
+        .ok(RoomMessagesResponseTemplate::default().end_token("further").events(vec![
+            f.text_msg("mine").sender(own_user_id).event_id(event_id!("$mine")),
+            f.text_msg("theirs").event_id(event_id!("$theirs")),
+            f.text_msg("first").event_id(event_id!("$first")),
+        ]))
+        .mock_once()
+        .mount()
+        .await;
+    let HistoryGapFillOutcome::Updated { gap, .. } =
+        cache.fill_history_gap_once(gaps[0].id.clone()).await.unwrap()
+    else {
+        panic!()
+    };
+    assert_eq!(gap.status, HistoryGapStatus::Joined);
+    let ids: Vec<_> =
+        cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();
     assert_eq!(
         ids,
-        vec![event_id!("$y"), event_id!("$z"), event_id!("$a"), event_id!("$live")]
+        vec![event_id!("$first"), event_id!("$theirs"), event_id!("$mine"), event_id!("$later")]
     );
 }

matrix-rust-sdk

Commit ba8ba5be174f

Commit `ba8ba5be174f`