matrix-rust-sdk

public

Star0

Fork 0

Watch0

44 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `dc40ad30111d`

tarball

atagen <boss@atagen.co>2026-10-27 00:36 UTC

3 files changed+149-0

event-cache: probe behind a dropped sync token for events the server skipped

A non-limited sync for a room with retained events drops its prev_batch, so a server that under-reports limited (seen with tuwunel's simplified sliding sync) loses whatever it skipped for good, with no gap to fill. The newest such token is now kept with its batch's first event, and probe_sync_hole fetches one page behind it through the history coordinator. Unknown events go in before that event, joining on any held event, or behind a history gap if the page never reaches one.

Modified file: crates/matrix-sdk/src/event_cache/caches/room/history.rs+61 additions@@ -99,6 +99,7 @@ enum HistoryOperation {
     Recent,
     Fill(OwnedTransactionId),
     SpendBudget(OwnedTransactionId),
+    ProbeSyncHole,
     Append(Vec<Event>, Option<String>),
     Restore,
     Barrier,
@@ -110,6 +111,7 @@ impl HistoryOperation {
             Self::Recent => "recent",
             Self::Fill(_) => "fill",
             Self::SpendBudget(_) => "spend budget",
+            Self::ProbeSyncHole => "probe sync hole",
             Self::Append(..) => "append",
             Self::Restore => "restore",
             Self::Barrier => "barrier",
@@ -121,6 +123,7 @@ impl HistoryOperation {
 enum HistoryResponse {
     Recent(RecentHistoryOutcome),
     Fill(HistoryGapFillOutcome),
+    Probe(Option<usize>),
     Append(Option<HistoryGap>),
     Restored,
 }
@@ -162,6 +165,9 @@ async fn coordinate_history(
                 HistoryOperation::SpendBudget(id) => {
                     cache.spend_history_budget_inner(id).await.map(HistoryResponse::Fill)
                 }
+                HistoryOperation::ProbeSyncHole => {
+                    cache.probe_sync_hole_inner().await.map(HistoryResponse::Probe)
+                }
                 HistoryOperation::Append(events, cursor) => cache
                     .append_history_section_inner(events, cursor)
                     .await
@@ -440,6 +446,61 @@ impl RoomEventCache {
         Ok((chunk.events().map(|(_, event)| event.clone()).collect(), gap_positions(chunk)))
     }
 
+    #[doc = "Looks once behind the newest previous-batch token a non-limited sync handed out. Answers None when there was none, else how many unknown events it inserted."]
+    pub async fn probe_sync_hole(&self) -> Result<Option<usize>> {
+        match self.request_history(HistoryOperation::ProbeSyncHole).await? {
+            HistoryResponse::Probe(outcome) => Ok(outcome),
+            _ => unreachable!("coordinator response matches request"),
+        }
+    }
+
+    async fn probe_sync_hole_inner(&self) -> Result<Option<usize>> {
+        let Some(hole) = self.inner.state.write().await?.take_sync_hole() else {
+            return Ok(None);
+        };
+        let room = self.inner.weak_room.get().ok_or(EventCacheError::ClientDropped)?;
+        let mut options = MessagesOptions::backward().from(hole.token.as_str());
+        options.limit = 50u16.into();
+        let response = room
+            .messages(options)
+            .await
+            .map_err(|error| EventCacheError::PaginationError(std::sync::Arc::new(error)))?;
+        let mut state = self.inner.state.write().await?;
+        let mut candidate = state.history_candidate().await?;
+        let known: HashSet<_> =
+            candidate.events().filter_map(|(_, event)| event.event_id()).collect();
+        let mut joined = false;
+        let mut missing = Vec::new();
+        for event in response.chunk {
+            match event.event_id() {
+                Some(id) if known.contains(&id) => {
+                    joined = true;
+                    break;
+                }
+                _ => missing.push(event),
+            }
+        }
+        let count = missing.len();
+        tracing::debug!(room_id = %self.room_id(), new = count, joined, "history gap probe");
+        if missing.is_empty() {
+            return Ok(Some(0));
+        }
+        missing.reverse();
+        let gap = (!joined).then(|| response.end).flatten().map(|token| Gap {
+            token,
+            history: Some(HistoryGap {
+                id: TransactionId::new(),
+                older_boundary: None,
+                newer_boundary: missing.iter().find_map(Event::event_id),
+                visited_cursors: Vec::new(),
+                status: HistoryGapStatus::Idle,
+            }),
+        });
+        candidate.insert_history_before(Some(&hole.before), missing.clone(), gap);
+        state.commit_history_candidate(candidate, missing, HistoryCommitScope::Loaded).await?;
+        Ok(Some(count))
+    }
+
     #[doc = "Whether any event is retained, in memory or in the store."]
     pub async fn has_retained_events(&self) -> Result<bool> {
         Ok(self.inner.state.write().await?.newest_retained_event_id().await?.is_some())

Modified file: crates/matrix-sdk/src/event_cache/caches/room/state.rs+23 additions@@ -151,6 +151,17 @@ pub struct RoomEventCacheState {
 
     /// A copy of the automatic pagination API object.
     automatic_pagination: Option<AutomaticPagination>,
+
+    /// The newest previous-batch token a non-limited sync handed out and we dropped, with the
+    /// first event of that batch. A server that under-reports `limited` hides skipped events
+    /// behind it, so [`super::RoomEventCache::probe_sync_hole`] looks once.
+    sync_hole: Option<SyncHole>,
+}
+
+#[derive(Clone, Debug)]
+pub(super) struct SyncHole {
+    pub token: String,
+    pub before: OwnedEventId,
 }
 
 impl RoomEventCacheState {
@@ -407,6 +418,7 @@ impl LockedRoomEventCacheState {
             linked_chunk_update_sender,
             room_version_rules,
             waited_for_initial_prev_token: false,
+            sync_hole: None,
             subscriber_count: Default::default(),
             pinned_event_cache: OnceLock::new(),
             automatic_pagination,
@@ -588,6 +600,10 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
         self.state.waited_for_initial_prev_token
     }
 
+    pub(super) fn take_sync_hole(&mut self) -> Option<SyncHole> {
+        self.state.sync_hole.take()
+    }
+
     /// Get a mutable reference to the `waited_for_initial_prev_token` value.
     pub fn waited_for_initial_prev_token_mut(&mut self) -> &mut bool {
         &mut self.state.waited_for_initial_prev_token
@@ -1025,6 +1041,13 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
         // knew about. This is slightly incorrect in the presence of
         // network splits, but this has shown to be Good Enough™.
         let newest_retained = self.newest_retained_event_id().await?;
+        if !timeline.limited && newest_retained.is_some() && !all_duplicates {
+            if let (Some(token), Some(before)) =
+                (prev_batch.clone(), events.iter().find_map(Event::event_id))
+            {
+                self.state.sync_hole = Some(SyncHole { token, before });
+            }
+        }
         if !timeline.limited && newest_retained.is_some() || all_duplicates {
             prev_batch = None;
         }

Modified file: crates/matrix-sdk/tests/recent_history.rs+65 additions@@ -708,3 +708,68 @@ async fn test_a_gap_whose_page_is_all_known_does_not_claim_the_start() {
     let outcome = cache.pagination().run_backwards_once(20).await.unwrap();
     assert!(outcome.reached_start);
 }
+
+#[async_test]
+async fn test_a_probe_behind_a_non_limited_sync_brings_back_what_it_skipped() {
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+    client.event_cache().subscribe().unwrap();
+    let room_id = room_id!("!hole: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"))),
+        )
+        .await;
+    assert_eq!(cache.probe_sync_hole().await.unwrap(), None);
+
+    // the server skipped $skipped but still said the timeline was not limited
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("new").event_id(event_id!("$new")))
+                .set_timeline_prev_batch("before-new"),
+        )
+        .await;
+    server
+        .mock_room_messages()
+        .match_from("before-new")
+        .ok(RoomMessagesResponseTemplate::default().end_token("further").events(vec![
+            f.text_msg("skipped").event_id(event_id!("$skipped")),
+            f.text_msg("held").event_id(event_id!("$held")),
+        ]))
+        .mock_once()
+        .mount()
+        .await;
+    assert_eq!(cache.probe_sync_hole().await.unwrap(), Some(1));
+    assert_eq!(cache.probe_sync_hole().await.unwrap(), None);
+    let ids: Vec<_> =
+        cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();
+    assert_eq!(ids, [event_id!("$held"), event_id!("$skipped"), event_id!("$new")]);
+    assert!(cache.history_gaps().await.unwrap().is_empty());
+
+    // an honest server's page starts at what we hold, and nothing changes
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("newer").event_id(event_id!("$newer")))
+                .set_timeline_prev_batch("before-newer"),
+        )
+        .await;
+    server
+        .mock_room_messages()
+        .match_from("before-newer")
+        .ok(RoomMessagesResponseTemplate::default()
+            .end_token("further")
+            .events(vec![f.text_msg("new").event_id(event_id!("$new"))]))
+        .mock_once()
+        .mount()
+        .await;
+    assert_eq!(cache.probe_sync_hole().await.unwrap(), Some(0));
+}

matrix-rust-sdk

Commit dc40ad30111d

Commit `dc40ad30111d`