matrix-rust-sdk

public

Star0

Fork 0

Watch0

38 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `ea600a4bd56a`

tarball

atagen <boss@atagen.co>2026-10-26 08:16 UTC

4 files changed+130-19

event-cache: resume back-pagination right before the oldest retained event

0f2f9829's fallback paginated from the live end, then put that page's token in front of the oldest retained event, where it led back into events already held. That misordered events, or claimed the start, or with a history gap present looped /messages without end, which wedged a device build after its first opens. Retained events with no gap before them now get their token from /context on the oldest one. Dropping a gap whose page was all known no longer counts as the start unless the server ran out or m.room.create is held.

Modified file: crates/matrix-sdk/src/event_cache/caches/room/history.rs+5 additions−4 deletions@@ -461,7 +461,9 @@ impl RoomEventCache {
             history.status = HistoryGapStatus::BudgetSpent;
             let history = history.clone();
             candidate.push_backwards_pagination_events(Some(chunk), Some(gap), &[]);
-            state.commit_history_candidate(candidate, Vec::new(), HistoryCommitScope::Loaded).await?;
+            state
+                .commit_history_candidate(candidate, Vec::new(), HistoryCommitScope::Loaded)
+                .await?;
             return Ok(HistoryGapFillOutcome::Updated { gap: history, events: 0 });
         }
         Ok(HistoryGapFillOutcome::Updated { gap: history.clone(), events: 0 })
@@ -685,9 +687,8 @@ fn apply_page(
         .flatten()
         .chain(history.older_boundary.clone())
         .collect();
-    let overlap = events
-        .iter()
-        .position(|event| event.event_id().is_some_and(|id| below.contains(&id)));
+    let overlap =
+        events.iter().position(|event| event.event_id().is_some_and(|id| below.contains(&id)));
     history.status = if let Some(message) = failure {
         HistoryGapStatus::Failed { message }
     } else if let Some(index) = overlap {

Modified file: crates/matrix-sdk/src/event_cache/caches/room/pagination.rs+53 additions−9 deletions@@ -201,20 +201,20 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
 
                 // a sync that drops its prev_batch (not limited, or every event a duplicate)
                 // leaves retained events with no gap before them, so only the creation event
-                // proves the start. otherwise paginate from the end, which dedups what we hold
-                let created = state.room_linked_chunk().events().next().is_some_and(|(_, event)| {
-                    event.raw().get_field::<String>("type").ok().flatten().as_deref()
-                        == Some("m.room.create")
-                });
-                if created {
+                // proves the start. otherwise the network side asks /context for a token just
+                // before the oldest event, and there is no sync token worth waiting for
+                let oldest = state.room_linked_chunk().events().next().map(|(_, event)| event);
+                if oldest.is_some_and(is_creation) {
                     trace!("chunk is fully loaded and starts at creation: reached_start=true");
                     return Ok(LoadMoreEventsBackwardsOutcome::StartOfTimeline);
                 }
+                let waited_for_initial_prev_token =
+                    oldest.is_some() || state.waited_for_initial_prev_token();
 
                 // Otherwise, start back-pagination from the end of the room.
                 return Ok(LoadMoreEventsBackwardsOutcome::Gap {
                     prev_token: None,
-                    waited_for_initial_prev_token: state.waited_for_initial_prev_token(),
+                    waited_for_initial_prev_token,
                 });
             }
 
@@ -301,7 +301,39 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
             return Ok(None);
         };
 
-        let mut options = MessagesOptions::new(Direction::Backward).from(prev_token.as_deref());
+        let from = match prev_token {
+            Some(token) => Some(token.clone()),
+            None => {
+                let oldest = self
+                    .state
+                    .read()
+                    .await?
+                    .room_linked_chunk()
+                    .events()
+                    .next()
+                    .and_then(|(_, event)| event.event_id());
+                match oldest {
+                    // a page from the live end would land in front of the oldest retained event
+                    // while belonging after it, so start right before that event instead
+                    Some(event_id) => {
+                        let mut request = ruma::api::client::context::get_context::v3::Request::new(
+                            self.room_id.clone(),
+                            event_id,
+                        );
+                        request.limit = ruma::uint!(0);
+                        let response = room.client().send(request).await.map_err(|err| {
+                            EventCacheError::PaginationError(Arc::new(err.into()))
+                        })?;
+                        match response.start {
+                            Some(start) => Some(start),
+                            None => return Ok(Some((Vec::new(), None))),
+                        }
+                    }
+                    None => None,
+                }
+            }
+        };
+        let mut options = MessagesOptions::new(Direction::Backward).from(from.as_deref());
         options.limit = batch_size.into();
 
         let response = room
@@ -355,6 +387,7 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
         }
 
         let from_end = prev_token.is_none();
+        let server_ended = new_token.is_none();
         // Check that the previous token still exists; otherwise it's a sign that the
         // room's timeline has been cleared.
         let prev_gap_id = if let Some(token) = prev_token {
@@ -428,11 +461,18 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
         let topo_ordered_events = events.iter().rev().cloned().collect::<Vec<_>>();
 
         let new_gap = new_token.map(|prev_token| Gap { token: prev_token, history: None });
+        // no gap left only proves the start when the server ran out or the room's creation is
+        // held. a token whose page we already held is dropped above without either
         let reached_start = state.room_linked_chunk_mut().push_backwards_pagination_events(
             prev_gap_id,
             new_gap,
             &topo_ordered_events,
-        );
+        ) && (server_ended
+            || state
+                .room_linked_chunk()
+                .events()
+                .next()
+                .is_some_and(|(_, event)| is_creation(event)));
 
         // A back-pagination can't include new read receipt events, as those are
         // ephemeral events not included in /messages responses, so we can
@@ -468,3 +508,7 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
         Ok(Some(BackPaginationOutcome { events, reached_start }))
     }
 }
+
+fn is_creation(event: &Event) -> bool {
+    event.raw().get_field::<String>("type").ok().flatten().as_deref() == Some("m.room.create")
+}

Modified file: crates/matrix-sdk/tests/integration/event_cache/mod.rs+2 additions−1 deletions@@ -1136,7 +1136,8 @@ async fn test_no_gap_stored_after_deduplicated_backpagination() {
     // Run pagination a second time: it will consume prev-batch, which is the least
     // recent token.
     let outcome = pagination.run_backwards_once(20).await.unwrap();
-    assert!(outcome.reached_start);
+    // the server still handed out prev-batch3, so dropping the useless gap is no start
+    assert!(!outcome.reached_start);
     assert!(outcome.events.is_empty());
     assert!(stream.is_empty());
 

Modified file: crates/matrix-sdk/tests/recent_history.rs+70 additions−5 deletions@@ -612,7 +612,9 @@ async fn test_recent_history_joins_on_any_retained_event() {
 }
 
 #[async_test]
-async fn test_retained_events_without_a_gap_paginate_from_the_end_until_creation() {
+async fn test_retained_events_without_a_gap_paginate_from_just_before_the_oldest() {
+    use matrix_sdk::test_utils::mocks::RoomContextResponseTemplate;
+
     let server = MatrixMockServer::new().await;
     let client = server.client_builder().build().await;
     client.event_cache().subscribe().unwrap();
@@ -628,11 +630,21 @@ async fn test_retained_events_without_a_gap_paginate_from_the_end_until_creation
         )
         .await;
     server
+        .mock_room_event_context()
+        .match_event_id()
+        .ok(RoomContextResponseTemplate::new(
+            f.text_msg("quiet").event_id(event_id!("$quiet")).into_event(),
+        )
+        .start("before-quiet"))
+        .mock_once()
+        .mount()
+        .await;
+    server
         .mock_room_messages()
-        .ok(RoomMessagesResponseTemplate::default().end_token("older").events(vec![
-            f.text_msg("quiet").event_id(event_id!("$quiet")),
-            f.text_msg("days ago").event_id(event_id!("$days-ago")),
-        ]))
+        .match_from("before-quiet")
+        .ok(RoomMessagesResponseTemplate::default()
+            .end_token("older")
+            .events(vec![f.text_msg("days ago").event_id(event_id!("$days-ago"))]))
         .mock_once()
         .mount()
         .await;
@@ -643,3 +655,56 @@ async fn test_retained_events_without_a_gap_paginate_from_the_end_until_creation
         cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();
     assert_eq!(ids, [event_id!("$days-ago"), event_id!("$quiet")]);
 }
+
+#[async_test]
+async fn test_a_gap_whose_page_is_all_known_does_not_claim_the_start() {
+    use matrix_sdk::test_utils::mocks::RoomContextResponseTemplate;
+
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+    client.event_cache().subscribe().unwrap();
+    let room_id = room_id!("!stale: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);
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("held").event_id(event_id!("$held")))
+                .set_timeline_limited()
+                .set_timeline_prev_batch("stale"),
+        )
+        .await;
+    server
+        .mock_room_messages()
+        .match_from("stale")
+        .ok(RoomMessagesResponseTemplate::default()
+            .end_token("past-held")
+            .events(vec![f.text_msg("held").event_id(event_id!("$held"))]))
+        .mock_once()
+        .mount()
+        .await;
+    let outcome = cache.pagination().run_backwards_once(20).await.unwrap();
+    assert!(!outcome.reached_start);
+
+    server
+        .mock_room_event_context()
+        .match_event_id()
+        .ok(RoomContextResponseTemplate::new(
+            f.text_msg("held").event_id(event_id!("$held")).into_event(),
+        )
+        .start("before-held"))
+        .mock_once()
+        .mount()
+        .await;
+    server
+        .mock_room_messages()
+        .match_from("before-held")
+        .ok(RoomMessagesResponseTemplate::default())
+        .mock_once()
+        .mount()
+        .await;
+    let outcome = cache.pagination().run_backwards_once(20).await.unwrap();
+    assert!(outcome.reached_start);
+}

matrix-rust-sdk

Commit ea600a4bd56a

Commit `ea600a4bd56a`