atagen

matrix-rust-sdk

public
Star0
Fork0
Watch0
35 commits4 branches0 tags0 releases16.6 MiB

Commit ec026bc9ea23

tarball

atagen <boss@atagen.co>2026-10-26 02:52 UTC

4 files changed+216-5
event-cache: join history on any event already below the gap

apply_page and the recent walk joined only on one exact event id, but /messages need not return that event (filtered, redacted, or ordered apart from sync arrival), so a gap over nothing missing paged through known history and stopped silently. A fill that runs out of pages can now be marked BudgetSpent, a new status so gaps stored as Paused still deserialise.
crates/matrix-sdk-base/src/event_cache/mod.rs+2
@@ -59,6 +59,8 @@ pub enum HistoryGapStatus {
5959 Idle,
6060 #[doc = "The server returned the same cursor, requiring an explicit retry."]
6161 Paused,
62+ #[doc = "A bounded fill ran out of pages before joining. Another fill continues from the cursor."]
63+ BudgetSpent,
6264 #[doc = "A failed request leaves the last successful cursor intact."]
6365 Failed {
6466 #[doc = "The request failure, retained for display after restart."]
crates/matrix-sdk/src/event_cache/caches/room/history.rs+74−5
@@ -98,6 +98,7 @@ pub(super) struct HistoryRequest {
9898 enum HistoryOperation {
9999 Recent,
100100 Fill(OwnedTransactionId),
101+ SpendBudget(OwnedTransactionId),
101102 Append(Vec<Event>, Option<String>),
102103 Restore,
103104 Barrier,
@@ -143,6 +144,9 @@ async fn coordinate_history(
143144 HistoryOperation::Fill(id) => {
144145 cache.fill_history_gap_inner(id, cancellation).await.map(HistoryResponse::Fill)
145146 }
147+ HistoryOperation::SpendBudget(id) => {
148+ cache.spend_history_budget_inner(id).await.map(HistoryResponse::Fill)
149+ }
146150 HistoryOperation::Append(events, cursor) => cache
147151 .append_history_section_inner(events, cursor)
148152 .await
@@ -232,7 +236,15 @@ impl RoomEventCache {
232236 &self,
233237 mut cancellation: watch::Receiver<bool>,
234238 ) -> Result<RecentHistoryOutcome> {
235- let anchor = self.inner.state.write().await?.newest_retained_event_id().await?;
239+ let (anchor, retained) = {
240+ let mut state = self.inner.state.write().await?;
241+ let retained: HashSet<_> = state
242+ .room_linked_chunk()
243+ .events()
244+ .filter_map(|(_, event)| event.event_id())
245+ .collect();
246+ (state.newest_retained_event_id().await?, retained)
247+ };
236248 let room = self.inner.weak_room.get().ok_or(EventCacheError::ClientDropped)?;
237249 let mut cursor = None;
238250 let mut visited = Vec::new();
@@ -258,7 +270,9 @@ impl RoomEventCache {
258270 raw_count += response.chunk.len();
259271 let mut page = response.chunk;
260272 if let Some(index) = anchor.as_ref().and_then(|anchor| {
261- page.iter().position(|event| event.event_id().as_ref() == Some(anchor))
273+ page.iter().position(|event| {
274+ event.event_id().is_some_and(|id| &id == anchor || retained.contains(&id))
275+ })
262276 }) {
263277 page.truncate(index);
264278 joined = true;
@@ -410,6 +424,49 @@ impl RoomEventCache {
410424 Ok((chunk.events().map(|(_, event)| event.clone()).collect(), gap_positions(chunk)))
411425 }
412426
427+ #[doc = "Whether any event is retained, in memory or in the store."]
428+ pub async fn has_retained_events(&self) -> Result<bool> {
429+ Ok(self.inner.state.write().await?.newest_retained_event_id().await?.is_some())
430+ }
431+
432+ #[doc = "Marks an idle gap whose bounded fill stopped before joining. A later fill clears it."]
433+ pub async fn spend_history_budget(
434+ &self,
435+ gap_id: OwnedTransactionId,
436+ ) -> Result<HistoryGapFillOutcome> {
437+ match self.request_history(HistoryOperation::SpendBudget(gap_id)).await? {
438+ HistoryResponse::Fill(outcome) => Ok(outcome),
439+ _ => unreachable!("coordinator response matches request"),
440+ }
441+ }
442+
443+ async fn spend_history_budget_inner(
444+ &self,
445+ gap_id: OwnedTransactionId,
446+ ) -> Result<HistoryGapFillOutcome> {
447+ let mut state = self.inner.state.write().await?;
448+ let mut candidate = state.history_candidate().await?;
449+ let Some((chunk, mut gap)) = candidate.chunks().find_map(|chunk| match chunk.content() {
450+ ChunkContent::Gap(gap)
451+ if gap.history.as_ref().is_some_and(|history| history.id == gap_id) =>
452+ {
453+ Some((chunk.identifier(), gap.clone()))
454+ }
455+ _ => None,
456+ }) else {
457+ return Ok(HistoryGapFillOutcome::NotFound);
458+ };
459+ let history = gap.history.as_mut().expect("matched a history gap");
460+ if history.status == HistoryGapStatus::Idle {
461+ history.status = HistoryGapStatus::BudgetSpent;
462+ let history = history.clone();
463+ candidate.push_backwards_pagination_events(Some(chunk), Some(gap), &[]);
464+ state.commit_history_candidate(candidate, Vec::new(), HistoryCommitScope::Loaded).await?;
465+ return Ok(HistoryGapFillOutcome::Updated { gap: history, events: 0 });
466+ }
467+ Ok(HistoryGapFillOutcome::Updated { gap: history.clone(), events: 0 })
468+ }
469+
413470 #[doc = "Includes terminal identities and gaps outside the currently loaded cache window."]
414471 pub async fn history_gaps(&self) -> Result<Vec<HistoryGap>> {
415472 let state = self.inner.state.write().await?;
@@ -616,9 +673,21 @@ fn apply_page(
616673 let ChunkContent::Gap(current) = gap_chunk.content() else { unreachable!() };
617674 let mut gap = current.clone();
618675 let history = gap.history.as_mut().expect("the gap was checked under the state lock");
619- let overlap = history.older_boundary.as_ref().and_then(|boundary| {
620- events.iter().position(|event| event.event_id().as_ref() == Some(boundary))
621- });
676+ // `/messages` may never return the boundary itself (filtered, redacted, or ordered apart
677+ // from sync arrival), so any event already below the gap proves the page reached it
678+ let below: HashSet<_> = candidate
679+ .chunks()
680+ .take_while(|chunk| chunk.identifier() != chunk_id)
681+ .filter_map(|chunk| match chunk.content() {
682+ ChunkContent::Items(items) => Some(items.iter().filter_map(Event::event_id)),
683+ ChunkContent::Gap(_) => None,
684+ })
685+ .flatten()
686+ .chain(history.older_boundary.clone())
687+ .collect();
688+ let overlap = events
689+ .iter()
690+ .position(|event| event.event_id().is_some_and(|id| below.contains(&id)));
622691 history.status = if let Some(message) = failure {
623692 HistoryGapStatus::Failed { message }
624693 } else if let Some(index) = overlap {
crates/matrix-sdk/tests/recent_history.rs+34
@@ -576,3 +576,37 @@ async fn test_joined_recent_history_leaves_no_gap_before_live_pagination() {
576576 cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();
577577 assert_eq!(ids, [event_id!("$e0"), event_id!("$e1"), event_id!("$e2")]);
578578 }
579+
580+#[async_test]
581+async fn test_recent_history_joins_on_any_retained_event() {
582+ let server = MatrixMockServer::new().await;
583+ let client = server.client_builder().build().await;
584+ client.event_cache().subscribe().unwrap();
585+ let room_id = room_id!("!history:example.org");
586+ let room = server.sync_joined_room(&client, room_id).await;
587+ let (cache, _handles) = room.event_cache().await.unwrap();
588+ let f = event_factory(room_id);
589+ server
590+ .sync_room(
591+ &client,
592+ JoinedRoomBuilder::new(room_id)
593+ .add_timeline_event(f.text_msg("older").event_id(event_id!("$older")))
594+ .add_timeline_event(f.text_msg("unlisted").event_id(event_id!("$unlisted"))),
595+ )
596+ .await;
597+ server
598+ .mock_room_messages()
599+ .match_limit(50)
600+ .ok(RoomMessagesResponseTemplate::default().end_token("unused").events(vec![
601+ f.text_msg("recent").event_id(event_id!("$recent")),
602+ f.text_msg("older").event_id(event_id!("$older")),
603+ ]))
604+ .mock_once()
605+ .mount()
606+ .await;
607+
608+ let outcome = cache.load_recent_history().await.unwrap();
609+ assert_eq!(outcome.pages, 1);
610+ assert_eq!(outcome.continuity, RecentHistoryContinuity::Joined);
611+ assert!(cache.history_gaps().await.unwrap().is_empty());
612+}
crates/matrix-sdk/tests/segmented_history.rs+106
@@ -664,3 +664,109 @@ async fn test_sync_superseding_a_trailing_gap_stops_its_pagination_cleanly() {
664664 assert_eq!(gaps.len(), 1);
665665 assert_eq!(gaps[0].older_boundary.as_deref(), Some(event_id!("$old")));
666666 }
667+
668+#[async_test]
669+async fn test_a_limited_sync_gap_joins_on_any_event_below_it() {
670+ let server = MatrixMockServer::new().await;
671+ let client = server.client_builder().build().await;
672+ client.event_cache().subscribe().unwrap();
673+ let room_id = room_id!("!history:example.org");
674+ let room = server.sync_joined_room(&client, room_id).await;
675+ let (cache, _handles) = room.event_cache().await.unwrap();
676+ let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
677+ server
678+ .sync_room(
679+ &client,
680+ JoinedRoomBuilder::new(room_id)
681+ .add_timeline_event(f.text_msg("older").event_id(event_id!("$older")))
682+ .add_timeline_event(f.text_msg("boundary").event_id(event_id!("$boundary"))),
683+ )
684+ .await;
685+ server
686+ .sync_room(
687+ &client,
688+ JoinedRoomBuilder::new(room_id)
689+ .add_timeline_event(f.text_msg("live").event_id(event_id!("$live")))
690+ .set_timeline_limited()
691+ .set_timeline_prev_batch("sync-edge"),
692+ )
693+ .await;
694+ let gaps = cache.history_gaps().await.unwrap();
695+ assert_eq!(gaps.len(), 1);
696+ assert_eq!(gaps[0].older_boundary.as_deref(), Some(event_id!("$boundary")));
697+
698+ // the server never hands back the boundary itself, only what lies below it
699+ server
700+ .mock_room_messages()
701+ .match_from("sync-edge")
702+ .match_limit(50)
703+ .ok(RoomMessagesResponseTemplate::default().end_token("further").events(vec![
704+ f.text_msg("missed").event_id(event_id!("$missed")),
705+ f.text_msg("older").event_id(event_id!("$older")),
706+ ]))
707+ .mock_once()
708+ .mount()
709+ .await;
710+ let HistoryGapFillOutcome::Updated { gap, events } =
711+ cache.fill_history_gap_once(gaps[0].id.clone()).await.unwrap()
712+ else {
713+ panic!()
714+ };
715+ assert_eq!(gap.status, HistoryGapStatus::Joined);
716+ assert_eq!(events, 1);
717+ assert!(cache.history_gaps().await.unwrap().is_empty());
718+ let ids: Vec<_> =
719+ cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();
720+ assert_eq!(
721+ ids,
722+ vec![event_id!("$older"), event_id!("$boundary"), event_id!("$missed"), event_id!("$live")]
723+ );
724+}
725+
726+#[async_test]
727+async fn test_a_spent_budget_marks_only_an_idle_gap_and_a_fill_clears_it() {
728+ let server = MatrixMockServer::new().await;
729+ let client = server.client_builder().build().await;
730+ client.event_cache().subscribe().unwrap();
731+ let room_id = room_id!("!history:example.org");
732+ let room = server.sync_joined_room(&client, room_id).await;
733+ let (cache, _handles) = room.event_cache().await.unwrap();
734+ let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
735+ server
736+ .sync_room(
737+ &client,
738+ JoinedRoomBuilder::new(room_id)
739+ .add_timeline_event(f.text_msg("retained").event_id(event_id!("$old"))),
740+ )
741+ .await;
742+ let gap = cache
743+ .append_history_section(
744+ vec![f.text_msg("recent").event_id(event_id!("$new")).into_event()],
745+ Some("recent-edge".into()),
746+ )
747+ .await
748+ .unwrap()
749+ .unwrap();
750+ let HistoryGapFillOutcome::Updated { gap: spent, .. } =
751+ cache.spend_history_budget(gap.id.clone()).await.unwrap()
752+ else {
753+ panic!()
754+ };
755+ assert_eq!(spent.status, HistoryGapStatus::BudgetSpent);
756+ assert_eq!(cache.history_gaps().await.unwrap()[0].status, HistoryGapStatus::BudgetSpent);
757+ server
758+ .mock_room_messages()
759+ .match_from("recent-edge")
760+ .ok(RoomMessagesResponseTemplate::default()
761+ .end_token("next")
762+ .events(vec![f.text_msg("middle").event_id(event_id!("$middle"))]))
763+ .mock_once()
764+ .mount()
765+ .await;
766+ let HistoryGapFillOutcome::Updated { gap: filled, .. } =
767+ cache.fill_history_gap_once(gap.id.clone()).await.unwrap()
768+ else {
769+ panic!()
770+ };
771+ assert_eq!(filled.status, HistoryGapStatus::Idle);
772+}