matrix-rust-sdk

public

Star0

Fork 0

Watch0

17 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `05ef287d1bfc`

tarball

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

8 files changed+238-54

event-cache: never leave adjacent or joined history gaps

A recent load captured its anchor before the request and slotted its window after any gap a limited sync installed meanwhile, leaving a stale gap row over continuous history in front of its own. The commit now drops that sync gap when the window overlaps the arrivals and otherwise moves it between the window and the arrivals, and a new live gap supersedes a trailing one. Joined gaps are no longer persisted by fills or recent loads, so every open stops leaving a gap chunk behind, and RecentHistoryOutcome reports a RecentHistoryContinuity instead of making callers read a gap status.

Modified file: crates/matrix-sdk-base/src/event_cache/mod.rs+1−1

@@ -66,6 +66,6 @@ pub enum HistoryGapStatus {

6666 },

6767 #[doc = "Accessible history ended without reaching the retained boundary."]

6868 Unavailable,

69- #[doc = "Pagination reached the retained boundary. The identity remains valid for stale callers."]

69+ #[doc = "Pagination reached the retained boundary. Only reported, a joined gap is removed."]

7070 Joined,

7171 }

Modified file: crates/matrix-sdk/src/event_cache/caches/event_linked_chunk.rs+15 additions@@ -155,6 +155,11 @@ impl EventLinkedChunk {
         Ok(())
     }
 
+    pub(super) fn remove_gap(&mut self, gap_identifier: ChunkIdentifier) {
+        self.replace_gap_at(gap_identifier, Vec::new())
+            .expect("gap_identifier is a valid gap chunk id we read previously");
+    }
+
     /// Remove some events from the linked chunk.
     ///
     /// If a chunk becomes empty, it's going to be removed.
@@ -320,9 +325,19 @@ impl EventLinkedChunk {
 
     /// Add the previous back-pagination token (if present), followed by the
     /// timeline events themselves.
+    ///
+    /// A new gap supersedes a trailing one, since it paginates over the same
+    /// interval from further ahead.
     pub fn push_live_events(&mut self, new_gap: Option<Gap>, events: &[Event]) {
         if let Some(new_gap) = new_gap {
             self.push_gap(new_gap);
+            let superseded = self
+                .rchunks()
+                .nth(1)
+                .and_then(|chunk| chunk.is_gap().then_some(chunk.identifier()));
+            if let Some(superseded) = superseded {
+                self.remove_gap(superseded);
+            }
         }
         self.chunks.push_items_back(events.iter().cloned());
     }

Modified file: crates/matrix-sdk/src/event_cache/caches/room/fill_history_gap_once.md+2 additions−2 deletions@@ -3,8 +3,8 @@ Ordinary oldest-history pagination and other internal gaps keep their cursors.
 
 Only reaching the recorded older boundary verifies a join. Short or empty pages
 with advancing cursors leave the gap idle. A repeated cursor pauses it. Exhausting
-accessible history leaves it unavailable and preserves retained events. Joined
-identities remain persisted so repeated calls can return their terminal outcome.
+accessible history leaves it unavailable and preserves retained events. A join
+removes the gap, so later calls with its identity return `NotFound`.
 
 Dropping the future before the response cancels the request without advancing
 the cursor. Once installation starts, the page and cursor commit together even

Modified file: crates/matrix-sdk/src/event_cache/caches/room/history.rs+95 additions−37 deletions@@ -52,14 +52,14 @@ pub(super) fn gap_positions(candidate: &EventLinkedChunk) -> Vec<HistoryGapPosit
 #[derive(Clone, Debug, PartialEq, Eq)]
 #[doc = "A bounded gap fill never falls back to another gap when its target disappears."]
 pub enum HistoryGapFillOutcome {
-    #[doc = "Also returned without a network request for a joined or unavailable interval."]
+    #[doc = "Also returned without a network request for an unavailable interval."]
     Updated {
         #[doc = "The durable outcome, including its original boundary identities."]
         gap: HistoryGap,
         #[doc = "Newly inserted events after room-wide deduplication."]
         events: usize,
     },
-    #[doc = "The identity is unknown or was removed by an explicit cache replacement."]
+    #[doc = "The identity is unknown, was joined, or was removed by a clear or a newer gap."]
     NotFound,
     #[doc = "Another history operation owns the room coordinator. No request was started."]
     Busy,
@@ -72,8 +72,19 @@ pub struct RecentHistoryOutcome {
     pub pages: u32,
     #[doc = "Number of raw events returned by the server."]
     pub events: usize,
-    #[doc = "The durable interval, absent when no retained boundary needs joining."]
-    pub gap: Option<HistoryGap>,
+    #[doc = "How the window relates to the retained history before it."]
+    pub continuity: RecentHistoryContinuity,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+#[doc = "Only a remaining interval is persisted, a join leaves no gap behind."]
+pub enum RecentHistoryContinuity {
+    #[doc = "No retained event preceded the window."]
+    Fresh,
+    #[doc = "The window, or sync arrivals during its request, reached the newest retained event."]
+    Joined,
+    #[doc = "The durable interval between the retained history and the window."]
+    Gap(HistoryGap),
 }
 
 pub(super) struct HistoryRequest {
@@ -267,42 +278,57 @@ impl RoomEventCache {
             let mut state = self.inner.state.write().await?;
             let mut candidate = state.history_candidate().await?;
             let baseline_count = candidate.events().count();
-            let fallback = {
-                let mut after_anchor = anchor.is_none();
-                candidate.events().find_map(|(_, event)| {
-                    let id = event.event_id();
-                    if after_anchor {
-                        return id;
+            let mut known = HashSet::new();
+            let mut arrivals = Vec::new();
+            let mut leading = Vec::new();
+            let mut passed_anchor = anchor.is_none();
+            for chunk in candidate.chunks() {
+                match chunk.content() {
+                    ChunkContent::Gap(gap) if passed_anchor && arrivals.is_empty() => {
+                        leading.push((chunk.identifier(), gap.clone()));
                     }
-                    if id == anchor {
-                        after_anchor = true;
+                    ChunkContent::Gap(_) => {}
+                    ChunkContent::Items(items) => {
+                        for id in items.iter().filter_map(Event::event_id) {
+                            if passed_anchor {
+                                arrivals.push(id.clone());
+                            }
+                            passed_anchor |= Some(&id) == anchor.as_ref();
+                            known.insert(id);
+                        }
                     }
-                    None
-                })
-            };
-            let newer_boundary = events.iter().find_map(Event::event_id).or(fallback.clone());
-            let history = anchor.map(|anchor| HistoryGap {
-                id: TransactionId::new(),
-                older_boundary: Some(anchor),
-                newer_boundary: newer_boundary.clone(),
-                status: if joined {
-                    HistoryGapStatus::Joined
-                } else if cursor.is_none() {
-                    HistoryGapStatus::Unavailable
-                } else if stalled {
-                    HistoryGapStatus::Paused
-                } else {
-                    HistoryGapStatus::Idle
-                },
-                visited_cursors: visited,
+                }
+            }
+            let anchor = anchor.filter(|_| passed_anchor);
+            if anchor.is_none() {
+                joined = false;
+            }
+            let first_arrival = arrivals.first().cloned();
+            let arrivals: HashSet<_> = arrivals.into_iter().collect();
+            let fetched = events.iter().any(|event| event.event_id().is_some());
+            let overlap = events
+                .iter()
+                .any(|event| event.event_id().is_some_and(|id| arrivals.contains(&id)));
+            let newest_fetched = events.iter().rev().find_map(Event::event_id);
+            let oldest_fetched = events.iter().find_map(Event::event_id);
+
+            // A gap that sync installed during the request survives only when the
+            // window ended before the sync arrivals, and then it belongs between them.
+            let mut kept = leading.pop_if(|(_, gap)| {
+                first_arrival.is_some()
+                    && !overlap
+                    && gap.history.as_ref().is_none_or(|h| h.status != HistoryGapStatus::Joined)
             });
-            let mut before = fallback;
+            let moved = kept.take_if(|_| fetched);
+            for (chunk, _) in leading.iter().chain(&moved) {
+                candidate.remove_gap(*chunk);
+            }
+
             let mut inserted = Vec::new();
+            let mut before = first_arrival.clone();
             for event in events.into_iter().rev() {
                 let id = event.event_id();
-                if id.as_ref().is_none_or(|id| {
-                    !candidate.events().any(|(_, event)| event.event_id().as_ref() == Some(id))
-                }) {
+                if id.as_ref().is_none_or(|id| known.insert(id.clone())) {
                     candidate.insert_history_before(before.as_deref(), vec![event.clone()], None);
                     inserted.push(event);
                 }
@@ -310,16 +336,47 @@ impl RoomEventCache {
                     before = id;
                 }
             }
-            if history.is_some() || cursor.is_some() {
+
+            let history = anchor.as_ref().filter(|_| !joined).map(|anchor| HistoryGap {
+                id: TransactionId::new(),
+                older_boundary: Some(anchor.clone()),
+                newer_boundary: oldest_fetched.or(first_arrival.clone()),
+                status: if cursor.is_none() {
+                    HistoryGapStatus::Unavailable
+                } else if stalled {
+                    HistoryGapStatus::Paused
+                } else {
+                    HistoryGapStatus::Idle
+                },
+                visited_cursors: visited,
+            });
+            let history = history.filter(|_| fetched || first_arrival.is_none());
+            if history.is_some() || cursor.is_some() && (fetched || first_arrival.is_none()) {
                 candidate.insert_history_before(
                     before.as_deref(),
                     Vec::new(),
                     Some(Gap { token: cursor.unwrap_or_default(), history: history.clone() }),
                 );
             }
+            if let Some((_, mut gap)) = moved {
+                if let Some(history) = gap.history.as_mut() {
+                    history.older_boundary = newest_fetched;
+                }
+                candidate.insert_history_before(first_arrival.as_deref(), Vec::new(), Some(gap));
+            }
+
+            let continuity = match (anchor, history, kept) {
+                (None, _, _) => RecentHistoryContinuity::Fresh,
+                (Some(_), Some(history), _) => RecentHistoryContinuity::Gap(history),
+                (Some(_), None, Some((_, Gap { history: Some(history), .. }))) => {
+                    RecentHistoryContinuity::Gap(history)
+                }
+                (Some(_), None, _) => RecentHistoryContinuity::Joined,
+            };
+
             inserted.reverse();
             state.commit_history_candidate(candidate, inserted, baseline_count).await?;
-            Ok(RecentHistoryOutcome { pages, events: raw_count, gap: history })
+            Ok(RecentHistoryOutcome { pages, events: raw_count, continuity })
         }
     }
 
@@ -546,6 +603,7 @@ fn apply_page(
     events.retain(|event| event.event_id().is_none_or(|id| known.insert(id)));
     events.reverse();
     let history = gap.history.as_ref().expect("history metadata was preserved").clone();
-    candidate.push_backwards_pagination_events(Some(chunk_id), Some(gap), &events);
+    let gap = (history.status != HistoryGapStatus::Joined).then_some(gap);
+    candidate.push_backwards_pagination_events(Some(chunk_id), gap, &events);
     (history, events)
 }

Modified file: crates/matrix-sdk/src/event_cache/caches/room/mod.rs+3 additions−1 deletions@@ -14,7 +14,9 @@
 
 mod history;
 pub mod pagination;
-pub use history::{HistoryGapFillOutcome, HistoryGapPosition, RecentHistoryOutcome};
+pub use history::{
+    HistoryGapFillOutcome, HistoryGapPosition, RecentHistoryContinuity, RecentHistoryOutcome,
+};
 mod state;
 mod subscriber;
 mod updates;

Modified file: crates/matrix-sdk/src/event_cache/mod.rs+3 additions−2 deletions@@ -70,8 +70,9 @@ pub use caches::{
     event_focused::EventFocusThreadMode,
     pagination::{BackPaginationOutcome, PaginationStatus},
     room::{
-        HistoryGapFillOutcome, HistoryGapPosition, RecentHistoryOutcome, RoomEventCache, RoomEventCacheGenericUpdate,
-        RoomEventCacheSubscriber, RoomEventCacheUpdate, pagination::RoomPagination,
+        HistoryGapFillOutcome, HistoryGapPosition, RecentHistoryContinuity, RecentHistoryOutcome,
+        RoomEventCache, RoomEventCacheGenericUpdate, RoomEventCacheSubscriber,
+        RoomEventCacheUpdate, pagination::RoomPagination,
     },
     thread::pagination::ThreadPagination,
 };

Modified file: crates/matrix-sdk/tests/recent_history.rs+117 additions−7 deletions@@ -1,7 +1,9 @@
 use std::time::Duration;
 
 use matrix_sdk::{
-    event_cache::HistoryGapStatus,
+    event_cache::{
+        HistoryGap, HistoryGapStatus, RecentHistoryContinuity, RecentHistoryOutcome, RoomEventCache,
+    },
     test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate},
 };
 use matrix_sdk_test::{JoinedRoomBuilder, async_test, event_factory::EventFactory};
@@ -11,6 +13,113 @@ fn event_factory(room_id: &ruma::RoomId) -> EventFactory {
     EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"))
 }
 
+fn gap(outcome: &RecentHistoryOutcome) -> &HistoryGap {
+    let RecentHistoryContinuity::Gap(gap) = &outcome.continuity else {
+        panic!("expected a remaining interval, got {:?}", outcome.continuity)
+    };
+    gap
+}
+
+async fn load_with_limited_sync_arrival(
+    server: &MatrixMockServer,
+    client: &matrix_sdk::Client,
+    cache: &RoomEventCache,
+    f: &EventFactory,
+) -> RecentHistoryOutcome {
+    let room_id = cache.room_id().to_owned();
+    let mut load = Box::pin(cache.load_recent_history());
+    tokio::select! {
+        biased;
+        result = &mut load => panic!("request finished before sync {result:?}"),
+        _ = tokio::time::sleep(Duration::from_millis(20)) => {}
+    }
+    server
+        .sync_room(
+            client,
+            JoinedRoomBuilder::new(&room_id)
+                .add_timeline_event(f.text_msg("live").event_id(event_id!("$live")))
+                .set_timeline_limited()
+                .set_timeline_prev_batch("sync-edge"),
+        )
+        .await;
+    load.await.unwrap()
+}
+
+#[async_test]
+async fn test_recent_history_overlapping_a_limited_sync_drops_its_stale_gap() {
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+    client.event_cache().subscribe().unwrap();
+    let room_id = room_id!("!history: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);
+    cache
+        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
+        .await
+        .unwrap();
+    server
+        .mock_room_messages()
+        .match_limit(50)
+        .ok(RoomMessagesResponseTemplate::default()
+            .events(vec![
+                f.text_msg("live").event_id(event_id!("$live")),
+                f.text_msg("recent").event_id(event_id!("$recent")),
+            ])
+            .with_delay(Duration::from_millis(100)))
+        .mock_once()
+        .mount()
+        .await;
+
+    let outcome = load_with_limited_sync_arrival(&server, &client, &cache, &f).await;
+
+    let (events, gaps) = cache.history_snapshot().await.unwrap();
+    let ids: Vec<_> = events.iter().filter_map(|event| event.event_id()).collect();
+    assert_eq!(ids, vec![event_id!("$old"), event_id!("$recent"), event_id!("$live")]);
+    assert_eq!(gaps.len(), 1);
+    assert_eq!(&gaps[0].gap, gap(&outcome));
+    assert_eq!(gaps[0].gap.older_boundary.as_deref(), Some(event_id!("$old")));
+    assert_eq!(gaps[0].next_event_id.as_deref(), Some(event_id!("$recent")));
+}
+
+#[async_test]
+async fn test_recent_history_ending_before_a_limited_sync_keeps_its_gap_after_the_window() {
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+    client.event_cache().subscribe().unwrap();
+    let room_id = room_id!("!history: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);
+    cache
+        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
+        .await
+        .unwrap();
+    server
+        .mock_room_messages()
+        .match_limit(50)
+        .ok(RoomMessagesResponseTemplate::default()
+            .events(vec![
+                f.text_msg("recent").event_id(event_id!("$recent")),
+                f.text_msg("old").event_id(event_id!("$old")),
+            ])
+            .with_delay(Duration::from_millis(100)))
+        .mock_once()
+        .mount()
+        .await;
+
+    let outcome = load_with_limited_sync_arrival(&server, &client, &cache, &f).await;
+    assert_eq!(outcome.continuity, RecentHistoryContinuity::Joined);
+
+    let (events, gaps) = cache.history_snapshot().await.unwrap();
+    let ids: Vec<_> = events.iter().filter_map(|event| event.event_id()).collect();
+    assert_eq!(ids, vec![event_id!("$old"), event_id!("$recent"), event_id!("$live")]);
+    assert_eq!(gaps.len(), 1);
+    assert_eq!(gaps[0].gap.older_boundary.as_deref(), Some(event_id!("$recent")));
+    assert_eq!(gaps[0].gap.newer_boundary.as_deref(), Some(event_id!("$live")));
+    assert_eq!(gaps[0].next_event_id.as_deref(), Some(event_id!("$live")));
+}
+
 #[async_test]
 async fn test_recent_history_is_bounded_and_keeps_the_retained_window() {
     let server = MatrixMockServer::new().await;
@@ -68,7 +177,7 @@ async fn test_recent_history_is_bounded_and_keeps_the_retained_window() {
     let outcome = cache.load_recent_history().await.unwrap();
     assert_eq!(outcome.pages, 4);
     assert_eq!(outcome.events, 4);
-    assert_eq!(outcome.gap.as_ref().unwrap().status, HistoryGapStatus::Idle);
+    assert_eq!(gap(&outcome).status, HistoryGapStatus::Idle);
     let ids: Vec<_> =
         cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();
     assert_eq!(
@@ -110,7 +219,8 @@ async fn test_recent_history_joins_at_the_retained_boundary() {
 
     let outcome = cache.load_recent_history().await.unwrap();
     assert_eq!(outcome.pages, 1);
-    assert_eq!(outcome.gap.as_ref().unwrap().status, HistoryGapStatus::Joined);
+    assert_eq!(outcome.continuity, RecentHistoryContinuity::Joined);
+    assert!(cache.history_gaps().await.unwrap().is_empty());
     assert_eq!(
         cache
             .events()
@@ -211,7 +321,7 @@ async fn test_recent_history_continues_empty_pages_and_pauses_on_a_cursor_cycle(
     let outcome = cache.load_recent_history().await.unwrap();
     assert_eq!(outcome.pages, 3);
     assert_eq!(outcome.events, 1);
-    assert_eq!(outcome.gap.as_ref().unwrap().status, HistoryGapStatus::Paused);
+    assert_eq!(gap(&outcome).status, HistoryGapStatus::Paused);
     assert_eq!(
         cache
             .events()
@@ -251,7 +361,7 @@ async fn test_concurrent_recent_history_calls_share_the_request_and_gap() {
     let (first, second) = tokio::join!(cache.load_recent_history(), cache.load_recent_history());
     let first = first.unwrap();
     let second = second.unwrap();
-    assert_eq!(first.gap.as_ref().unwrap().id, second.gap.as_ref().unwrap().id);
+    assert_eq!(gap(&first).id, gap(&second).id);
     assert_eq!(first.pages, 1);
     assert_eq!(second.pages, 1);
     assert_eq!(first.events, 1);
@@ -294,7 +404,7 @@ async fn test_cancelling_one_recent_history_waiter_keeps_the_shared_request_aliv
     drop(first);
     let outcome = second.await.unwrap().unwrap();
     assert_eq!(outcome.pages, 1);
-    assert_eq!(outcome.gap.unwrap().status, HistoryGapStatus::Unavailable);
+    assert_eq!(gap(&outcome).status, HistoryGapStatus::Unavailable);
     assert_eq!(
         cache
             .events()
@@ -361,7 +471,7 @@ async fn test_cancelling_all_recent_history_waiters_allows_a_fresh_retry() {
 
     let outcome = cache.load_recent_history().await.unwrap();
     assert_eq!(outcome.pages, 1);
-    assert_eq!(outcome.gap.unwrap().status, HistoryGapStatus::Unavailable);
+    assert_eq!(gap(&outcome).status, HistoryGapStatus::Unavailable);
     assert_eq!(
         cache
             .events()

Modified file: crates/matrix-sdk/tests/segmented_history.rs+2 additions−4 deletions@@ -113,10 +113,8 @@ async fn test_segmented_history_resumes_after_restart_and_joins_only_at_boundary
     };
     assert_eq!(gap.status, HistoryGapStatus::Joined);
     assert_eq!(events, 0);
-    assert_eq!(
-        cache.fill_history_gap_once(gap_id).await.unwrap(),
-        HistoryGapFillOutcome::Updated { gap, events: 0 }
-    );
+    assert!(cache.history_gaps().await.unwrap().is_empty());
+    assert_eq!(cache.fill_history_gap_once(gap_id).await.unwrap(), HistoryGapFillOutcome::NotFound);
     let ids: Vec<_> =
         cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();
     assert_eq!(ids, vec![event_id!("$old"), event_id!("$middle"), event_id!("$new")]);

matrix-rust-sdk

Commit 05ef287d1bfc

Commit `05ef287d1bfc`