atagen

matrix-rust-sdk

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

Commit 49a9cd1aff19

tarball

atagen <boss@atagen.co>2026-10-25 07:04 UTC

41 files changed+2570-131
event-cache: preserve segmented room history
benchmarks/benches/linked_chunk.rs+5−3
@@ -83,8 +83,10 @@ fn writing(c: &mut Criterion) {
8383 }
8484
8585 {
86- operations
87- .push(Operation::PushGapBack(Gap { token: format!("gap{gap_nth}") }));
86+ operations.push(Operation::PushGapBack(Gap {
87+ token: format!("gap{gap_nth}"),
88+ history: None,
89+ }));
8890 gap_nth += 1;
8991 }
9092 }
@@ -187,7 +189,7 @@ fn reading(c: &mut Criterion) {
187189 }
188190
189191 lc.push_items_back(events_chunk);
190- lc.push_gap_back(Gap { token: format!("gap{num_gaps}") });
192+ lc.push_gap_back(Gap { token: format!("gap{num_gaps}"), history: None });
191193
192194 num_gaps += 1;
193195 }
bindings/matrix-sdk-ffi/src/timeline/mod.rs+32
@@ -914,6 +914,22 @@ impl TimelineItem {
914914 use matrix_sdk_ui::timeline::VirtualTimelineItem as VItem;
915915 match self.0.as_virtual()? {
916916 VItem::DateDivider(ts) => Some(VirtualTimelineItem::DateDivider { ts: (*ts).into() }),
917+ VItem::HistoryGap(gap) => Some(VirtualTimelineItem::HistoryGap {
918+ id: gap.id.to_string(),
919+ older_boundary: gap.older_boundary.as_ref().map(ToString::to_string),
920+ newer_boundary: gap.newer_boundary.as_ref().map(ToString::to_string),
921+ status: match &gap.status {
922+ matrix_sdk::event_cache::HistoryGapStatus::Idle => HistoryGapStatus::Idle,
923+ matrix_sdk::event_cache::HistoryGapStatus::Paused => HistoryGapStatus::Paused,
924+ matrix_sdk::event_cache::HistoryGapStatus::Failed { message } => {
925+ HistoryGapStatus::Failed { message: message.clone() }
926+ }
927+ matrix_sdk::event_cache::HistoryGapStatus::Unavailable => {
928+ HistoryGapStatus::Unavailable
929+ }
930+ matrix_sdk::event_cache::HistoryGapStatus::Joined => HistoryGapStatus::Joined,
931+ },
932+ }),
917933 VItem::ReadMarker => Some(VirtualTimelineItem::ReadMarker),
918934 VItem::TimelineStart => Some(VirtualTimelineItem::TimelineStart),
919935 }
@@ -1205,6 +1221,13 @@ pub enum VirtualTimelineItem {
12051221 ts: Timestamp,
12061222 },
12071223
1224+ HistoryGap {
1225+ id: String,
1226+ older_boundary: Option<String>,
1227+ newer_boundary: Option<String>,
1228+ status: HistoryGapStatus,
1229+ },
1230+
12081231 /// The user's own read marker.
12091232 ReadMarker,
12101233
@@ -1212,6 +1235,15 @@ pub enum VirtualTimelineItem {
12121235 TimelineStart,
12131236 }
12141237
1238+#[derive(Clone, uniffi::Enum)]
1239+pub enum HistoryGapStatus {
1240+ Idle,
1241+ Paused,
1242+ Failed { message: String },
1243+ Unavailable,
1244+ Joined,
1245+}
1246+
12151247 /// A [`TimelineItem`](super::TimelineItem) that doesn't correspond to an event.
12161248 #[derive(uniffi::Enum)]
12171249 pub enum ReceiptType {
crates/matrix-sdk-base/src/event_cache/mod.rs+40−1
@@ -15,6 +15,8 @@
1515 //! Event cache store and common types shared with `matrix_sdk::event_cache`.
1616
1717 use matrix_sdk_common::deserialized_responses::TimelineEvent;
18+use ruma::{OwnedEventId, OwnedTransactionId};
19+use serde::{Deserialize, Serialize};
1820
1921 pub mod store;
2022
@@ -22,11 +24,48 @@ pub mod store;
2224 pub type Event = TimelineEvent;
2325
2426 /// The kind of gap the event storage holds.
25-#[derive(Clone, Debug)]
27+#[derive(Clone, Debug, Serialize, Deserialize)]
2628 pub struct Gap {
2729 /// The token to use in the pagination query as the `from` parameter,
2830 /// extracted from a previous `start` / `end` field of a `/messages`
2931 /// response, or from the `prev_batch` / `next_batch` field of a `/sync`
3032 /// response.
3133 pub token: String,
34+ #[serde(default, skip_serializing_if = "Option::is_none")]
35+ #[doc = "Present for an interval that must survive independently of ordinary oldest-history pagination."]
36+ pub history: Option<HistoryGap>,
37+}
38+
39+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
40+#[doc = "Durable identity and boundary evidence for an independently pageable interval."]
41+pub struct HistoryGap {
42+ #[doc = "Remains unchanged when pagination replaces the underlying gap chunk."]
43+ pub id: OwnedTransactionId,
44+ #[doc = "Pagination must reach this retained event before claiming a verified join."]
45+ pub older_boundary: Option<OwnedEventId>,
46+ #[doc = "The original event on the newer side, preserved as pages are inserted."]
47+ pub newer_boundary: Option<OwnedEventId>,
48+ #[doc = "Previously consumed cursors detect server pagination cycles across restarts."]
49+ #[serde(default, skip_serializing_if = "Vec::is_empty")]
50+ pub visited_cursors: Vec<String>,
51+ #[doc = "Unavailable and joined intervals remain distinguishable after restart."]
52+ pub status: HistoryGapStatus,
53+}
54+
55+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
56+#[doc = "Persisted outcomes distinguish server exhaustion from verified overlap."]
57+pub enum HistoryGapStatus {
58+ #[doc = "Another bounded request can continue from the saved cursor."]
59+ Idle,
60+ #[doc = "The server returned the same cursor, requiring an explicit retry."]
61+ Paused,
62+ #[doc = "A failed request leaves the last successful cursor intact."]
63+ Failed {
64+ #[doc = "The request failure, retained for display after restart."]
65+ message: String,
66+ },
67+ #[doc = "Accessible history ended without reaching the retained boundary."]
68+ Unavailable,
69+ #[doc = "Pagination reached the retained boundary. The identity remains valid for stale callers."]
70+ Joined,
3271 }
crates/matrix-sdk-base/src/event_cache/store/integration_tests.rs+125−13
@@ -42,7 +42,7 @@ use ruma::{
4242 };
4343
4444 use super::DynEventCacheStore;
45-use crate::event_cache::{Gap, store::DEFAULT_CHUNK_CAPACITY};
45+use crate::event_cache::{Gap, HistoryGap, HistoryGapStatus, store::DEFAULT_CHUNK_CAPACITY};
4646
4747 /// Create a test event with all data filled, for testing that linked chunk
4848 /// correctly stores event data.
@@ -137,6 +137,9 @@ pub trait EventCacheStoreIntegrationTests {
137137 /// the store.
138138 async fn test_handle_updates_and_rebuild_linked_chunk(&self);
139139
140+ #[doc = "Gap identity and cursor advancement must share the transaction that inserts their page."]
141+ async fn test_history_gap_roundtrip_and_replacement(&self);
142+
140143 /// Test that the next and previous fields only reference chunks that
141144 /// already exist in the store.
142145 async fn test_linked_chunk_exists_before_referenced(&self);
@@ -268,7 +271,7 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
268271 previous: Some(CId::new(0)),
269272 new: CId::new(1),
270273 next: None,
271- gap: Gap { token: "parmesan".to_owned() },
274+ gap: Gap { token: "parmesan".to_owned(), history: None },
272275 },
273276 // another items chunk
274277 Update::NewItemsChunk { previous: Some(CId::new(1)), new: CId::new(2), next: None },
@@ -326,6 +329,108 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
326329 assert!(chunks.next().is_none());
327330 }
328331
332+ async fn test_history_gap_roundtrip_and_replacement(&self) {
333+ let room_id = room_id!("!r0:matrix.org");
334+ let linked_chunk_id = LinkedChunkId::Room(room_id);
335+ let history = HistoryGap {
336+ id: "history-gap-id".into(),
337+ older_boundary: Some(event_id!("$older-boundary").to_owned()),
338+ newer_boundary: Some(event_id!("$newer-boundary").to_owned()),
339+ visited_cursors: Vec::new(),
340+ status: HistoryGapStatus::Failed { message: "retry later".to_owned() },
341+ };
342+ let expected_history = history.clone();
343+
344+ self.handle_linked_chunk_updates(
345+ linked_chunk_id,
346+ vec![
347+ Update::NewItemsChunk { previous: None, new: CId::new(0), next: None },
348+ Update::PushItems {
349+ at: Position::new(CId::new(0), 0),
350+ items: vec![make_test_event(room_id, "before")],
351+ },
352+ Update::NewGapChunk {
353+ previous: Some(CId::new(0)),
354+ new: CId::new(1),
355+ next: None,
356+ gap: Gap { token: "old history token".to_owned(), history: Some(history) },
357+ },
358+ Update::NewItemsChunk { previous: Some(CId::new(1)), new: CId::new(2), next: None },
359+ Update::PushItems {
360+ at: Position::new(CId::new(2), 0),
361+ items: vec![make_test_event(room_id, "after")],
362+ },
363+ Update::NewGapChunk {
364+ previous: Some(CId::new(2)),
365+ new: CId::new(3),
366+ next: None,
367+ gap: Gap { token: "other gap token".to_owned(), history: None },
368+ },
369+ ],
370+ )
371+ .await
372+ .unwrap();
373+
374+ self.handle_linked_chunk_updates(
375+ linked_chunk_id,
376+ vec![
377+ Update::RemoveChunk(CId::new(1)),
378+ Update::NewItemsChunk {
379+ previous: Some(CId::new(0)),
380+ new: CId::new(4),
381+ next: Some(CId::new(2)),
382+ },
383+ Update::PushItems {
384+ at: Position::new(CId::new(4), 0),
385+ items: vec![
386+ make_test_event(room_id, "inserted first"),
387+ make_test_event(room_id, "inserted second"),
388+ ],
389+ },
390+ Update::NewGapChunk {
391+ previous: Some(CId::new(4)),
392+ new: CId::new(5),
393+ next: Some(CId::new(2)),
394+ gap: Gap {
395+ token: "new history token".to_owned(),
396+ history: Some(expected_history.clone()),
397+ },
398+ },
399+ ],
400+ )
401+ .await
402+ .unwrap();
403+
404+ let linked_chunk = lazy_loader::from_all_chunks::<DEFAULT_CHUNK_CAPACITY, _, _>(
405+ self.load_all_chunks(linked_chunk_id).await.unwrap(),
406+ )
407+ .unwrap()
408+ .unwrap();
409+ let chunks: Vec<_> = linked_chunk.chunks().collect();
410+ assert_eq!(chunks.len(), 5);
411+ assert_matches!(chunks[0].content(), ChunkContent::Items(events) => {
412+ assert_eq!(events.len(), 1);
413+ check_test_event(&events[0], "before");
414+ });
415+ assert_matches!(chunks[1].content(), ChunkContent::Items(events) => {
416+ assert_eq!(events.len(), 2);
417+ check_test_event(&events[0], "inserted first");
418+ check_test_event(&events[1], "inserted second");
419+ });
420+ assert_matches!(chunks[2].content(), ChunkContent::Gap(gap) => {
421+ assert_eq!(gap.token, "new history token");
422+ assert_eq!(gap.history.as_ref(), Some(&expected_history));
423+ });
424+ assert_matches!(chunks[3].content(), ChunkContent::Items(events) => {
425+ assert_eq!(events.len(), 1);
426+ check_test_event(&events[0], "after");
427+ });
428+ assert_matches!(chunks[4].content(), ChunkContent::Gap(gap) => {
429+ assert_eq!(gap.token, "other gap token");
430+ assert!(gap.history.is_none());
431+ });
432+ }
433+
329434 async fn test_linked_chunk_exists_before_referenced(&self) {
330435 let room_id = *DEFAULT_TEST_ROOM_ID;
331436 let linked_chunk_id = LinkedChunkId::Room(room_id);
@@ -361,7 +466,7 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
361466 previous: Some(CId::new(41)),
362467 new: CId::new(42),
363468 next: None,
364- gap: Gap { token: "gap".to_owned() },
469+ gap: Gap { token: "gap".to_owned(), history: None },
365470 }],
366471 )
367472 .await
@@ -374,7 +479,7 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
374479 previous: None,
375480 new: CId::new(42),
376481 next: Some(CId::new(43)),
377- gap: Gap { token: "gap".to_owned() },
482+ gap: Gap { token: "gap".to_owned(), history: None },
378483 }],
379484 )
380485 .await
@@ -459,7 +564,7 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
459564 previous: Some(CId::new(0)),
460565 new: CId::new(1),
461566 next: None,
462- gap: Gap { token: "parmesan".to_owned() },
567+ gap: Gap { token: "parmesan".to_owned(), history: None },
463568 },
464569 // another items chunk
465570 Update::NewItemsChunk { previous: Some(CId::new(1)), new: CId::new(2), next: None },
@@ -708,7 +813,7 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
708813 previous: Some(CId::new(0)),
709814 new: CId::new(1),
710815 next: None,
711- gap: Gap { token: "morbier".to_owned() },
816+ gap: Gap { token: "morbier".to_owned(), history: None },
712817 },
713818 // new chunk for items
714819 Update::NewItemsChunk { previous: Some(CId::new(1)), new: CId::new(2), next: None },
@@ -907,19 +1012,19 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
9071012 previous: None,
9081013 new: CId::new(42),
9091014 next: None,
910- gap: Gap { token: "raclette".to_owned() },
1015+ gap: Gap { token: "raclette".to_owned(), history: None },
9111016 },
9121017 Update::NewGapChunk {
9131018 previous: Some(CId::new(42)),
9141019 new: CId::new(43),
9151020 next: None,
916- gap: Gap { token: "fondue".to_owned() },
1021+ gap: Gap { token: "fondue".to_owned(), history: None },
9171022 },
9181023 Update::NewGapChunk {
9191024 previous: Some(CId::new(43)),
9201025 new: CId::new(44),
9211026 next: None,
922- gap: Gap { token: "tartiflette".to_owned() },
1027+ gap: Gap { token: "tartiflette".to_owned(), history: None },
9231028 },
9241029 Update::RemoveChunk(CId::new(43)),
9251030 ],
@@ -1125,7 +1230,7 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
11251230 previous: Some(CId::new(42)),
11261231 new: CId::new(54),
11271232 next: None,
1128- gap: Gap { token: "fondue".to_owned() },
1233+ gap: Gap { token: "fondue".to_owned(), history: None },
11291234 },
11301235 Update::PushItems {
11311236 at: Position::new(CId::new(42), 0),
@@ -1156,7 +1261,7 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
11561261 previous: Some(CId::new(42)),
11571262 new: CId::new(54),
11581263 next: None,
1159- gap: Gap { token: "fondue".to_owned() },
1264+ gap: Gap { token: "fondue".to_owned(), history: None },
11601265 },
11611266 Update::PushItems {
11621267 at: Position::new(CId::new(42), 0),
@@ -1285,7 +1390,7 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
12851390 previous: Some(CId::new(0)),
12861391 new: CId::new(1),
12871392 next: None,
1288- gap: Gap { token: "bleu d'auvergne".to_owned() },
1393+ gap: Gap { token: "bleu d'auvergne".to_owned(), history: None },
12891394 },
12901395 // another items chunk
12911396 Update::NewItemsChunk { previous: Some(CId::new(1)), new: CId::new(2), next: None },
@@ -1578,7 +1683,7 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
15781683 previous: Some(CId::new(0)),
15791684 new: CId::new(1),
15801685 next: None,
1581- gap: Gap { token: "brillat-savarin".to_owned() },
1686+ gap: Gap { token: "brillat-savarin".to_owned(), history: None },
15821687 },
15831688 Update::NewItemsChunk { previous: Some(CId::new(1)), new: CId::new(2), next: None },
15841689 Update::PushItems {
@@ -2483,6 +2588,13 @@ macro_rules! event_cache_store_integration_tests {
24832588 }
24842589
24852590 #[async_test]
2591+ async fn test_history_gap_roundtrip_and_replacement() {
2592+ let event_cache_store =
2593+ get_event_cache_store().await.unwrap().into_event_cache_store();
2594+ event_cache_store.test_history_gap_roundtrip_and_replacement().await;
2595+ }
2596+
2597+ #[async_test]
24862598 async fn test_linked_chunk_exists_before_referenced() {
24872599 let event_cache_store =
24882600 get_event_cache_store().await.unwrap().into_event_cache_store();
crates/matrix-sdk-indexeddb/src/event_cache_store/mod.rs+1
@@ -196,6 +196,7 @@ impl EventCacheStore for IndexeddbEventCacheStore {
196196 linked_chunk_id: linked_chunk_id.to_owned(),
197197 chunk_identifier: new.index(),
198198 token: gap.token,
199+ history: gap.history,
199200 })
200201 .await?;
201202 transaction
crates/matrix-sdk-indexeddb/src/event_cache_store/transaction.rs+1−1
@@ -236,7 +236,7 @@ impl<'a> IndexeddbEventCacheStoreTransaction<'a> {
236236 .get_gap_by_id(linked_chunk_id, ChunkIdentifier::new(chunk.identifier))
237237 .await?
238238 .ok_or(TransactionError::ItemNotFound)?;
239- ChunkContent::Gap(RawGap { token: gap.token })
239+ ChunkContent::Gap(RawGap { token: gap.token, history: gap.history })
240240 }
241241 };
242242 return Ok(Some(RawChunk {
crates/matrix-sdk-indexeddb/src/event_cache_store/types.rs+2
@@ -231,4 +231,6 @@ pub struct Gap {
231231 /// "end" field of a `/messages` response.
232232 #[serde(alias = "prev_token")]
233233 pub token: String,
234+ #[serde(default, skip_serializing_if = "Option::is_none")]
235+ pub history: Option<matrix_sdk_base::event_cache::HistoryGap>,
234236 }
crates/matrix-sdk-sqlite/src/event_cache_store.rs+17−5
@@ -385,8 +385,16 @@ impl TransactionExtForLinkedChunks for Transaction<'_> {
385385 |row| row.get(0),
386386 )?;
387387 let prev_token_bytes = store.decode_value(&encoded_prev_token)?;
388- let prev_token = serde_json::from_slice(&prev_token_bytes)?;
389- Ok(Gap { token: prev_token })
388+ #[derive(serde::Deserialize)]
389+ #[serde(untagged)]
390+ enum StoredGap {
391+ Legacy(String),
392+ History(Gap),
393+ }
394+ Ok(match serde_json::from_slice(&prev_token_bytes)? {
395+ StoredGap::Legacy(token) => Gap { token, history: None },
396+ StoredGap::History(gap) => gap,
397+ })
390398 }
391399
392400 fn load_events_content(
@@ -689,7 +697,11 @@ impl EventCacheStore for SqliteEventCacheStore {
689697 }
690698
691699 Update::NewGapChunk { previous, new, next, gap } => {
692- let serialized = serde_json::to_vec(&gap.token)?;
700+ let serialized = if gap.history.is_some() {
701+ serde_json::to_vec(&gap)?
702+ } else {
703+ serde_json::to_vec(&gap.token)?
704+ };
693705 let prev_token = this.encode_value(serialized)?;
694706
695707 let previous = previous.as_ref().map(ChunkIdentifier::index);
@@ -1981,7 +1993,7 @@ mod tests {
19811993 previous: None,
19821994 new: ChunkIdentifier::new(chunk_id),
19831995 next: None,
1984- gap: Gap { token: "gap".to_owned() },
1996+ gap: Gap { token: "gap".to_owned(), history: None },
19851997 }],
19861998 )
19871999 .await
@@ -2081,7 +2093,7 @@ mod tests {
20812093 previous: None,
20822094 new: ChunkIdentifier::new(2),
20832095 next: None,
2084- gap: Gap { token: "new gap".to_owned() },
2096+ gap: Gap { token: "new gap".to_owned(), history: None },
20852097 }],
20862098 )
20872099 .await
crates/matrix-sdk-ui/src/timeline/builder.rs+19−3
@@ -17,6 +17,7 @@ use std::sync::Arc;
1717 use matrix_sdk::Room;
1818 use matrix_sdk_base::{SendOutsideWasm, SyncOutsideWasm};
1919 use ruma::{events::AnySyncTimelineEvent, room_version_rules::RoomVersionRules};
20+use tokio::sync::mpsc;
2021 use tracing::{Instrument, Span, info_span};
2122
2223 use super::{
@@ -162,7 +163,11 @@ impl TimelineBuilder {
162163 room.client().event_cache().subscribe()?;
163164
164165 let (room_event_cache, event_cache_drop) = room.event_cache().await?;
165- let (_, event_subscriber) = room_event_cache.subscribe().await?;
166+ if matches!(focus, TimelineFocus::Live { .. }) {
167+ room_event_cache.load_retained_history().await?;
168+ }
169+ let (initial_events, initial_gaps, event_subscriber) =
170+ room_event_cache.subscribe_with_history().await?;
166171
167172 let is_room_encrypted = room
168173 .latest_encryption_state()
@@ -180,8 +185,17 @@ impl TimelineBuilder {
180185 settings,
181186 );
182187
183- let InitFocusResult { focus_task, has_events } =
184- controller.init_focus(&focus, &room_event_cache).await?;
188+ let InitFocusResult { focus_task, has_events } = controller
189+ .init_focus(&focus, &room_event_cache, (initial_events, initial_gaps))
190+ .await?;
191+
192+ let (history_flush, history_flush_requests) = if matches!(focus, TimelineFocus::Live { .. })
193+ {
194+ let (sender, receiver) = mpsc::channel(1);
195+ (Some(sender), Some(receiver))
196+ } else {
197+ (None, None)
198+ };
185199
186200 let room_update_join_handle = room
187201 .client()
@@ -201,6 +215,7 @@ impl TimelineBuilder {
201215 controller.clone(),
202216 event_subscriber,
203217 focus.clone(),
218+ history_flush_requests,
204219 )
205220 .instrument(span)
206221 })
@@ -238,6 +253,7 @@ impl TimelineBuilder {
238253 let timeline = Timeline {
239254 controller,
240255 event_cache: room_event_cache,
256+ history_flush,
241257 drop_handle: Arc::new(TimelineDropHandle {
242258 _crypto_drop_handles: crypto_drop_handles,
243259 _room_update_join_handle: room_update_join_handle,
crates/matrix-sdk-ui/src/timeline/controller/metadata.rs+3
@@ -59,6 +59,7 @@ pub(crate) struct RemoteEventContext<'a> {
5959
6060 #[derive(Clone, Debug)]
6161 pub(in crate::timeline) struct TimelineMetadata {
62+ pub history_gaps: Vec<matrix_sdk::event_cache::HistoryGapPosition>,
6263 // **** CONSTANT FIELDS ****
6364 /// An optional prefix for internal IDs, defined during construction of the
6465 /// timeline.
@@ -136,6 +137,7 @@ impl TimelineMetadata {
136137 is_room_encrypted: bool,
137138 ) -> Self {
138139 Self {
140+ history_gaps: Vec::new(),
139141 subscriber_skip_count: SkipCount::new(),
140142 own_user_id,
141143 next_internal_id: Default::default(),
@@ -154,6 +156,7 @@ impl TimelineMetadata {
154156 }
155157
156158 pub(super) fn clear(&mut self) {
159+ self.history_gaps.clear();
157160 // Note: we don't clear the next internal id to avoid bad cases of stale unique
158161 // ids across timeline clears.
159162 self.aggregations.clear();
crates/matrix-sdk-ui/src/timeline/controller/mod.rs+37−2
@@ -668,6 +668,27 @@ impl<P: RoomDataProvider> TimelineController<P> {
668668 .await
669669 }
670670
671+ pub(super) async fn handle_history_update(
672+ &self,
673+ diffs: Vec<VectorDiff<TimelineEvent>>,
674+ gaps: Option<Vec<matrix_sdk::event_cache::HistoryGapPosition>>,
675+ origin: RemoteEventOrigin,
676+ ) {
677+ let mut state = self.state.write().await;
678+ let mut txn = state.transaction();
679+ txn.handle_remote_events_with_diffs(
680+ diffs,
681+ origin,
682+ &self.room_data_provider,
683+ &self.settings,
684+ )
685+ .await;
686+ if let Some(gaps) = gaps {
687+ txn.meta.history_gaps = gaps;
688+ }
689+ txn.commit();
690+ }
691+
671692 /// Only handle aggregations received as [`VectorDiff`]s.
672693 pub(super) async fn handle_remote_aggregations(
673694 &self,
@@ -703,6 +724,18 @@ impl<P: RoomDataProvider> TimelineController<P> {
703724 Events: IntoIterator,
704725 <Events as IntoIterator>::Item: Into<TimelineEvent>,
705726 {
727+ self.replace_with_initial_history(events, Vec::new(), origin).await;
728+ }
729+
730+ pub(super) async fn replace_with_initial_history<Events>(
731+ &self,
732+ events: Events,
733+ gaps: Vec<matrix_sdk::event_cache::HistoryGapPosition>,
734+ origin: RemoteEventOrigin,
735+ ) where
736+ Events: IntoIterator,
737+ <Events as IntoIterator>::Item: Into<TimelineEvent>,
738+ {
706739 let mut state = self.state.write().await;
707740
708741 let track_read_markers = &self.settings.track_read_receipts;
@@ -723,6 +756,7 @@ impl<P: RoomDataProvider> TimelineController<P> {
723756 state
724757 .replace_with_remote_events(
725758 events,
759+ gaps,
726760 origin,
727761 &self.room_data_provider,
728762 &self.settings,
@@ -1374,15 +1408,16 @@ impl TimelineController {
13741408 &self,
13751409 focus: &TimelineFocus,
13761410 room_event_cache: &RoomEventCache,
1411+ initial_history: (Vec<TimelineEvent>, Vec<matrix_sdk::event_cache::HistoryGapPosition>),
13771412 ) -> Result<InitFocusResult, Error> {
13781413 match focus {
13791414 TimelineFocus::Live { .. } => {
13801415 // Retrieve the cached events, and add them to the timeline.
1381- let events = room_event_cache.events().await?;
1416+ let (events, gaps) = initial_history;
13821417
13831418 let has_events = !events.is_empty();
13841419
1385- self.replace_with_initial_remote_events(events, RemoteEventOrigin::Cache).await;
1420+ self.replace_with_initial_history(events, gaps, RemoteEventOrigin::Cache).await;
13861421
13871422 match room_event_cache.pagination().status().get() {
13881423 PaginationStatus::Idle { hit_timeline_start } => {
crates/matrix-sdk-ui/src/timeline/controller/state.rs+2
@@ -230,6 +230,7 @@ impl<P: RoomDataProvider> TimelineState<P> {
230230 pub(super) async fn replace_with_remote_events<Events>(
231231 &mut self,
232232 events: Events,
233+ gaps: Vec<matrix_sdk::event_cache::HistoryGapPosition>,
233234 origin: RemoteEventOrigin,
234235 room_data_provider: &P,
235236 settings: &TimelineSettings,
@@ -246,6 +247,7 @@ impl<P: RoomDataProvider> TimelineState<P> {
246247 settings,
247248 )
248249 .await;
250+ txn.meta.history_gaps = gaps;
249251 txn.commit();
250252 }
251253
crates/matrix-sdk-ui/src/timeline/controller/state_transaction.rs+78−9
@@ -1029,9 +1029,9 @@ impl<'a, P: RoomDataProvider> TimelineStateTransaction<'a, P> {
10291029 if entry.is_remote_event()
10301030 || entry.as_virtual().is_some_and(|vitem| match vitem {
10311031 VirtualTimelineItem::DateDivider(_) => false,
1032- VirtualTimelineItem::ReadMarker | VirtualTimelineItem::TimelineStart => {
1033- true
1034- }
1032+ VirtualTimelineItem::HistoryGap(_)
1033+ | VirtualTimelineItem::ReadMarker
1034+ | VirtualTimelineItem::TimelineStart => true,
10351035 })
10361036 {
10371037 ObservableItemsTransactionEntry::remove(entry);
@@ -1070,16 +1070,85 @@ impl<'a, P: RoomDataProvider> TimelineStateTransaction<'a, P> {
10701070 self.meta.update_read_marker(&mut self.items);
10711071 }
10721072
1073- pub(super) fn commit(self) {
1073+ fn reconcile_history_gaps(&mut self) {
1074+ for index in (0..self.items.len()).rev() {
1075+ if let Some(VirtualTimelineItem::HistoryGap(gap)) = self.items[index].as_virtual() {
1076+ if !self.meta.history_gaps.iter().any(|position| {
1077+ position.gap.id == gap.id
1078+ && !matches!(
1079+ position.gap.status,
1080+ matrix_sdk::event_cache::HistoryGapStatus::Joined
1081+ )
1082+ }) {
1083+ self.items.remove(index);
1084+ }
1085+ }
1086+ }
1087+ for position in self.meta.history_gaps.clone() {
1088+ if matches!(position.gap.status, matrix_sdk::event_cache::HistoryGapStatus::Joined) {
1089+ continue;
1090+ }
1091+ let mut index = if let Some(next) = &position.next_event_id {
1092+ let mut events = self.items.all_remote_events().iter();
1093+ if !events.any(|event| &event.event_id == next) {
1094+ continue;
1095+ }
1096+ self.items
1097+ .all_remote_events()
1098+ .get_by_event_id(next)
1099+ .and_then(|event| event.timeline_item_index)
1100+ .or_else(|| events.find_map(|event| event.timeline_item_index))
1101+ } else {
1102+ None
1103+ };
1104+ let target = index.get_or_insert_with(|| {
1105+ self.items
1106+ .iter_all_regions()
1107+ .find_map(|(index, item)| item.is_local_echo().then_some(index))
1108+ .unwrap_or(self.items.len())
1109+ });
1110+ let previous = self.items.iter_all_regions().find_map(|(index, item)| {
1111+ matches!(item.as_virtual(), Some(VirtualTimelineItem::HistoryGap(gap)) if gap.id == position.gap.id)
1112+ .then_some((index, item.clone()))
1113+ });
1114+ if let Some((index, item)) = &previous {
1115+ if *index + 1 == *target
1116+ && matches!(item.as_virtual(), Some(VirtualTimelineItem::HistoryGap(gap)) if gap == &position.gap)
1117+ {
1118+ continue;
1119+ }
1120+ }
1121+ let kind = VirtualTimelineItem::HistoryGap(position.gap.clone());
1122+ let item = if let Some((index, item)) = previous {
1123+ self.items.remove(index);
1124+ if index < *target {
1125+ *target -= 1;
1126+ }
1127+ item.with_kind(kind)
1128+ } else {
1129+ self.meta.new_timeline_item(kind)
1130+ };
1131+ self.items.insert(*target, item, None);
1132+ }
1133+ }
1134+
1135+ pub(super) fn commit(mut self) {
1136+ self.reconcile_history_gaps();
10741137 // Update the `subscriber_skip_count` value.
10751138 let previous_number_of_items = self.number_of_items_when_transaction_started;
10761139 let next_number_of_items = self.items.len();
10771140
1078- if previous_number_of_items != next_number_of_items {
1079- let count = self
1080- .meta
1081- .subscriber_skip_count
1082- .compute_next(previous_number_of_items, next_number_of_items);
1141+ let has_history_gaps = self.meta.history_gaps.iter().any(|position| {
1142+ !matches!(position.gap.status, matrix_sdk::event_cache::HistoryGapStatus::Joined)
1143+ });
1144+ if previous_number_of_items != next_number_of_items || has_history_gaps {
1145+ let count = if has_history_gaps {
1146+ 0
1147+ } else {
1148+ self.meta
1149+ .subscriber_skip_count
1150+ .compute_next(previous_number_of_items, next_number_of_items)
1151+ };
10831152 self.meta
10841153 .subscriber_skip_count
10851154 .update(count, matches!(self.focus, TimelineFocusKind::Live { .. }));
crates/matrix-sdk-ui/src/timeline/date_dividers.rs+6−3
@@ -154,7 +154,8 @@ impl DateDividerAdjuster {
154154 }
155155
156156 TimelineItemKind::Virtual(VirtualTimelineItem::ReadMarker)
157- | TimelineItemKind::Virtual(VirtualTimelineItem::TimelineStart) => {
157+ | TimelineItemKind::Virtual(VirtualTimelineItem::TimelineStart)
158+ | TimelineItemKind::Virtual(VirtualTimelineItem::HistoryGap(_)) => {
158159 // Nothing to do.
159160 }
160161 }
@@ -251,7 +252,8 @@ impl DateDividerAdjuster {
251252 }
252253
253254 TimelineItemKind::Virtual(VirtualTimelineItem::ReadMarker)
254- | TimelineItemKind::Virtual(VirtualTimelineItem::TimelineStart) => {
255+ | TimelineItemKind::Virtual(VirtualTimelineItem::TimelineStart)
256+ | TimelineItemKind::Virtual(VirtualTimelineItem::HistoryGap(_)) => {
255257 // Nothing to do.
256258 }
257259 }
@@ -317,7 +319,8 @@ impl DateDividerAdjuster {
317319 }
318320
319321 TimelineItemKind::Virtual(VirtualTimelineItem::ReadMarker)
320- | TimelineItemKind::Virtual(VirtualTimelineItem::TimelineStart) => {
322+ | TimelineItemKind::Virtual(VirtualTimelineItem::TimelineStart)
323+ | TimelineItemKind::Virtual(VirtualTimelineItem::HistoryGap(_)) => {
321324 // Nothing to do.
322325 }
323326 }
crates/matrix-sdk-ui/src/timeline/mod.rs+16
@@ -56,6 +56,7 @@ use ruma::{
5656 };
5757 use subscriber::TimelineWithDropHandle;
5858 use thiserror::Error;
59+use tokio::sync::{mpsc, oneshot};
5960 use tracing::{instrument, trace, warn};
6061
6162 use self::{
@@ -118,6 +119,8 @@ pub struct Timeline {
118119 /// The event cache specialized for this room's view.
119120 event_cache: RoomEventCache,
120121
122+ history_flush: Option<mpsc::Sender<oneshot::Sender<Result<(), Error>>>>,
123+
121124 /// References to long-running tasks held by the timeline.
122125 drop_handle: Arc<TimelineDropHandle>,
123126 }
@@ -224,6 +227,19 @@ pub struct AttachmentConfig {
224227 }
225228
226229 impl Timeline {
230+ #[doc = "Wait until this live timeline has applied all history currently retained by its room cache."]
231+ pub async fn flush_history(&self) -> Result<(), Error> {
232+ let Some(flush) = &self.history_flush else {
233+ return Ok(());
234+ };
235+ let (reply, result) = oneshot::channel();
236+ flush.send(reply).await.map_err(|_| {
237+ Error::EventCacheError(matrix_sdk::event_cache::EventCacheError::ClientDropped)
238+ })?;
239+ result.await.map_err(|_| {
240+ Error::EventCacheError(matrix_sdk::event_cache::EventCacheError::ClientDropped)
241+ })?
242+ }
227243 /// Returns the room for this timeline.
228244 pub fn room(&self) -> &Room {
229245 self.controller.room()
crates/matrix-sdk-ui/src/timeline/tasks.rs+72−16
@@ -16,6 +16,7 @@
1616
1717 use std::collections::BTreeSet;
1818
19+use eyeball_im::VectorDiff;
1920 use matrix_sdk::{
2021 event_cache::{
2122 EventFocusThreadMode, EventsOrigin, RoomEventCache, RoomEventCacheSubscriber,
@@ -25,6 +26,7 @@ use matrix_sdk::{
2526 };
2627 use ruma::OwnedEventId;
2728 use tokio::sync::broadcast::{Receiver, error::RecvError};
29+use tokio::sync::{mpsc, oneshot};
2830 use tracing::{error, instrument, trace, warn};
2931
3032 use crate::timeline::{TimelineController, TimelineFocus, event_item::RemoteEventOrigin};
@@ -203,13 +205,47 @@ pub(in crate::timeline) async fn room_event_cache_updates_task(
203205 timeline_controller: TimelineController,
204206 mut room_event_cache_subscriber: RoomEventCacheSubscriber,
205207 timeline_focus: TimelineFocus,
208+ mut history_flush_requests: Option<
209+ mpsc::Receiver<oneshot::Sender<Result<(), crate::timeline::Error>>>,
210+ >,
206211 ) {
207212 trace!("Spawned the event subscriber task.");
208213
209214 loop {
210215 trace!("Waiting for an event.");
211216
212- let update = match room_event_cache_subscriber.recv().await {
217+ let update = tokio::select! {
218+ request = async {
219+ match history_flush_requests.as_mut() {
220+ Some(requests) => requests.recv().await,
221+ None => std::future::pending::<Option<oneshot::Sender<Result<(), crate::timeline::Error>>>>().await,
222+ }
223+ } => {
224+ if let Some(reply) = request {
225+ if let Err(error) = room_event_cache.wait_for_history().await {
226+ let _ = reply.send(Err(error.into()));
227+ continue;
228+ }
229+ match room_event_cache.subscribe_with_history().await {
230+ Ok((initial_events, gaps, subscriber)) => {
231+ room_event_cache_subscriber = subscriber;
232+ timeline_controller
233+ .replace_with_initial_history(initial_events, gaps, RemoteEventOrigin::Cache)
234+ .await;
235+ let _ = reply.send(Ok(()));
236+ }
237+ Err(error) => {
238+ let _ = reply.send(Err(error.into()));
239+ }
240+ }
241+ } else {
242+ history_flush_requests = None;
243+ }
244+ continue;
245+ }
246+ update = room_event_cache_subscriber.recv() => update,
247+ };
248+ let update = match update {
213249 Ok(up) => up,
214250 Err(RecvError::Closed) => break,
215251 Err(RecvError::Lagged(num_skipped)) => {
@@ -218,20 +254,36 @@ pub(in crate::timeline) async fn room_event_cache_updates_task(
218254 // The updates might have lagged, but the room event cache might have
219255 // events, so retrieve them and add them back again to the timeline,
220256 // after clearing it.
221- let initial_events = match room_event_cache.events().await {
222- Ok(initial_events) => initial_events,
223- Err(err) => {
224- error!(
225- ?err,
226- "Failed to replace the initial remote events in the event cache"
227- );
228- break;
229- }
230- };
257+ let (initial_events, gaps, subscriber) =
258+ match room_event_cache.subscribe_with_history().await {
259+ Ok(initial_events) => initial_events,
260+ Err(err) => {
261+ error!(
262+ ?err,
263+ "Failed to replace the initial remote events in the event cache"
264+ );
265+ break;
266+ }
267+ };
268+
269+ room_event_cache_subscriber = subscriber;
231270
232- timeline_controller
233- .replace_with_initial_remote_events(initial_events, RemoteEventOrigin::Cache)
234- .await;
271+ if matches!(timeline_focus, TimelineFocus::Live { .. }) {
272+ timeline_controller
273+ .replace_with_initial_history(
274+ initial_events,
275+ gaps,
276+ RemoteEventOrigin::Cache,
277+ )
278+ .await;
279+ } else if !matches!(timeline_focus, TimelineFocus::PinnedEvents) {
280+ timeline_controller
281+ .handle_remote_aggregations(
282+ vec![VectorDiff::Reset { values: initial_events.into() }],
283+ RemoteEventOrigin::Cache,
284+ )
285+ .await;
286+ }
235287
236288 continue;
237289 }
@@ -243,7 +295,11 @@ pub(in crate::timeline) async fn room_event_cache_updates_task(
243295 timeline_controller.handle_fully_read_marker(event_id).await;
244296 }
245297
246- RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs, origin }) => {
298+ RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
299+ diffs,
300+ origin,
301+ history_gaps,
302+ }) => {
247303 trace!("Received new timeline events diffs");
248304 let origin = match origin {
249305 EventsOrigin::Sync => RemoteEventOrigin::Sync,
@@ -254,7 +310,7 @@ pub(in crate::timeline) async fn room_event_cache_updates_task(
254310 let has_diffs = !diffs.is_empty();
255311
256312 if matches!(timeline_focus, TimelineFocus::Live { .. }) {
257- timeline_controller.handle_remote_events_with_diffs(diffs, origin).await;
313+ timeline_controller.handle_history_update(diffs, history_gaps, origin).await;
258314 } else if !matches!(timeline_focus, TimelineFocus::PinnedEvents) {
259315 // Only handle the remote aggregation for a non-live timeline, that's not the
260316 // pinned events one (since the latter handles remote aggregations on its own).
crates/matrix-sdk-ui/src/timeline/tests/mod.rs+1
@@ -66,6 +66,7 @@ mod polls;
6666 mod reactions;
6767 mod read_receipts;
6868 mod redaction;
69+mod segmented_history;
6970 mod shields;
7071 mod virt;
7172
crates/matrix-sdk-ui/src/timeline/tests/segmented_history.rs+228
@@ -0,0 +1,228 @@
1+use eyeball_im::VectorDiff;
2+use futures_util::{FutureExt, StreamExt};
3+use imbl::vector;
4+use matrix_sdk::event_cache::{HistoryGap, HistoryGapPosition, HistoryGapStatus};
5+use matrix_sdk_test::{ALICE, async_test};
6+use ruma::{
7+ TransactionId, event_id,
8+ events::{AnyMessageLikeEventContent, room::message::RoomMessageEventContent},
9+};
10+
11+use super::TestTimeline;
12+use crate::timeline::{VirtualTimelineItem, event_item::RemoteEventOrigin};
13+
14+#[async_test]
15+async fn test_segmented_history_initial_snapshot_keeps_retained_side() {
16+ let timeline = TestTimeline::new();
17+ let events: Vec<matrix_sdk::deserialized_responses::TimelineEvent> = (0..35)
18+ .map(|_| timeline.factory.text_msg("retained or recent").sender(*ALICE).into())
19+ .collect();
20+ let old = events[0].event_id().unwrap();
21+ let new = events[1].event_id().unwrap();
22+ let gap = HistoryGap {
23+ id: TransactionId::new(),
24+ older_boundary: Some(old.clone()),
25+ newer_boundary: Some(new.clone()),
26+ visited_cursors: Vec::new(),
27+ status: HistoryGapStatus::Idle,
28+ };
29+ timeline
30+ .controller
31+ .replace_with_initial_history(
32+ events,
33+ vec![HistoryGapPosition { gap, next_event_id: Some(new) }],
34+ RemoteEventOrigin::Cache,
35+ )
36+ .await;
37+ let (items, _) = timeline.controller.subscribe().await;
38+ assert_eq!(items.iter().filter(|item| item.is_remote_event()).count(), 35);
39+ assert!(
40+ items
41+ .iter()
42+ .any(|item| item.as_event().is_some_and(|event| event.event_id() == Some(&old)))
43+ );
44+ assert!(
45+ items
46+ .iter()
47+ .any(|item| matches!(item.as_virtual(), Some(VirtualTimelineItem::HistoryGap(_))))
48+ );
49+}
50+
51+#[async_test]
52+async fn test_segmented_history_projects_hidden_boundaries_atomically() {
53+ let timeline = TestTimeline::new();
54+ let (mut items, mut stream) = timeline.controller.subscribe().await;
55+ let old = event_id!("$old");
56+ let hidden = event_id!("$reaction");
57+ let new = event_id!("$new");
58+ let gap = HistoryGap {
59+ id: TransactionId::new(),
60+ older_boundary: Some(old.to_owned()),
61+ newer_boundary: Some(hidden.to_owned()),
62+ visited_cursors: Vec::new(),
63+ status: HistoryGapStatus::Idle,
64+ };
65+ let f = &timeline.factory;
66+ timeline
67+ .controller
68+ .handle_history_update(
69+ vec![VectorDiff::Append {
70+ values: vector![
71+ f.text_msg("old").sender(*ALICE).event_id(old).into(),
72+ f.reaction(old, "👍").sender(*ALICE).event_id(hidden).into(),
73+ f.text_msg("new").sender(*ALICE).event_id(new).into(),
74+ ],
75+ }],
76+ Some(vec![HistoryGapPosition {
77+ gap: gap.clone(),
78+ next_event_id: Some(hidden.to_owned()),
79+ }]),
80+ RemoteEventOrigin::Cache,
81+ )
82+ .await;
83+ let diffs = stream.next().await.unwrap();
84+ for diff in diffs {
85+ diff.apply(&mut items);
86+ }
87+ let gap_index = items.iter().position(|item| matches!(item.as_virtual(), Some(VirtualTimelineItem::HistoryGap(found)) if found == &gap)).unwrap();
88+ assert_eq!(items[gap_index - 1].as_event().unwrap().event_id(), Some(old));
89+ assert_eq!(items[gap_index + 1].as_event().unwrap().event_id(), Some(new));
90+ assert!(stream.next().now_or_never().is_none());
91+
92+ let gap_key = items[gap_index].unique_id().clone();
93+ let mut paused = gap.clone();
94+ paused.status = HistoryGapStatus::Paused;
95+ timeline
96+ .controller
97+ .handle_history_update(
98+ Vec::new(),
99+ Some(vec![HistoryGapPosition {
100+ gap: paused.clone(),
101+ next_event_id: Some(hidden.to_owned()),
102+ }]),
103+ RemoteEventOrigin::Cache,
104+ )
105+ .await;
106+ for diff in stream.next().await.unwrap() {
107+ diff.apply(&mut items);
108+ }
109+ assert_eq!(items[gap_index].unique_id(), &gap_key);
110+ assert!(
111+ matches!(items[gap_index].as_virtual(), Some(VirtualTimelineItem::HistoryGap(found)) if found == &paused)
112+ );
113+
114+ let old_key = items[gap_index - 1].unique_id().clone();
115+ let new_key = items[gap_index + 1].unique_id().clone();
116+ let middle = event_id!("$middle");
117+ timeline
118+ .controller
119+ .handle_history_update(
120+ vec![VectorDiff::Insert {
121+ index: 1,
122+ value: f.text_msg("middle").sender(*ALICE).event_id(middle).into(),
123+ }],
124+ Some(vec![HistoryGapPosition {
125+ gap: paused.clone(),
126+ next_event_id: Some(middle.to_owned()),
127+ }]),
128+ RemoteEventOrigin::Cache,
129+ )
130+ .await;
131+ for diff in stream.next().await.unwrap() {
132+ diff.apply(&mut items);
133+ }
134+ assert_eq!(
135+ items
136+ .iter()
137+ .find(|item| item.as_event().is_some_and(|event| event.event_id() == Some(old)))
138+ .unwrap()
139+ .unique_id(),
140+ &old_key
141+ );
142+ assert_eq!(
143+ items
144+ .iter()
145+ .find(|item| item.as_event().is_some_and(|event| event.event_id() == Some(new)))
146+ .unwrap()
147+ .unique_id(),
148+ &new_key
149+ );
150+ assert_eq!(
151+ items
152+ .iter()
153+ .find(|item| matches!(item.as_virtual(), Some(VirtualTimelineItem::HistoryGap(_))))
154+ .unwrap()
155+ .unique_id(),
156+ &gap_key
157+ );
158+
159+ paused.status = HistoryGapStatus::Joined;
160+ timeline
161+ .controller
162+ .handle_history_update(
163+ Vec::new(),
164+ Some(vec![HistoryGapPosition { gap: paused, next_event_id: Some(hidden.to_owned()) }]),
165+ RemoteEventOrigin::Cache,
166+ )
167+ .await;
168+ for diff in stream.next().await.unwrap() {
169+ diff.apply(&mut items);
170+ }
171+ assert!(
172+ !items
173+ .iter()
174+ .any(|item| matches!(item.as_virtual(), Some(VirtualTimelineItem::HistoryGap(_))))
175+ );
176+}
177+
178+#[async_test]
179+async fn test_segmented_history_reset_preserves_pending_sends() {
180+ let timeline = TestTimeline::new();
181+ timeline
182+ .handle_local_event(AnyMessageLikeEventContent::RoomMessage(
183+ RoomMessageEventContent::text_plain("pending"),
184+ ))
185+ .await;
186+ let pending_key = timeline
187+ .controller
188+ .items()
189+ .await
190+ .iter()
191+ .find(|item| item.is_local_echo())
192+ .unwrap()
193+ .unique_id()
194+ .clone();
195+ let old = event_id!("$old");
196+ let new = event_id!("$new");
197+ let gap = HistoryGap {
198+ id: TransactionId::new(),
199+ older_boundary: Some(old.to_owned()),
200+ newer_boundary: Some(new.to_owned()),
201+ visited_cursors: Vec::new(),
202+ status: HistoryGapStatus::Unavailable,
203+ };
204+ let f = &timeline.factory;
205+ timeline
206+ .controller
207+ .handle_history_update(
208+ vec![VectorDiff::Reset {
209+ values: vector![
210+ f.text_msg("old").sender(*ALICE).event_id(old).into(),
211+ f.text_msg("new").sender(*ALICE).event_id(new).into(),
212+ ],
213+ }],
214+ Some(vec![HistoryGapPosition { gap, next_event_id: Some(new.to_owned()) }]),
215+ RemoteEventOrigin::Cache,
216+ )
217+ .await;
218+ let items = timeline.controller.items().await;
219+ assert_eq!(items.iter().filter(|item| item.is_local_echo()).count(), 1);
220+ assert_eq!(items.iter().find(|item| item.is_local_echo()).unwrap().unique_id(), &pending_key);
221+ assert_eq!(
222+ items
223+ .iter()
224+ .filter(|item| matches!(item.as_virtual(), Some(VirtualTimelineItem::HistoryGap(_))))
225+ .count(),
226+ 1
227+ );
228+}
crates/matrix-sdk-ui/src/timeline/virtual_item.rs+2
@@ -17,6 +17,8 @@ use ruma::MilliSecondsSinceUnixEpoch;
1717 /// A [`TimelineItem`](super::TimelineItem) that doesn't correspond to an event.
1818 #[derive(Clone, Debug)]
1919 pub enum VirtualTimelineItem {
20+ #[doc = "An unresolved interval between retained history sections."]
21+ HistoryGap(matrix_sdk::event_cache::HistoryGap),
2022 /// A divider between messages of two days or months depending on the
2123 /// timeline configuration.
2224 ///
crates/matrix-sdk-ui/tests/integration/timeline/mod.rs+1
@@ -57,6 +57,7 @@ mod reactions;
5757 mod read_receipts;
5858 mod redecryption;
5959 mod reseed;
60+mod segmented_history;
6061 mod replies;
6162 mod subscribe;
6263 mod thread;
crates/matrix-sdk-ui/tests/integration/timeline/segmented_history.rs+156
@@ -0,0 +1,156 @@
1+use std::time::Duration;
2+
3+use futures_util::StreamExt;
4+use matrix_sdk::test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate};
5+use matrix_sdk_test::{ALICE, async_test, event_factory::EventFactory};
6+use matrix_sdk_ui::timeline::{RoomExt, VirtualTimelineItem};
7+use ruma::{event_id, events::room::message::RoomMessageEventContent, room_id};
8+use tokio::time::timeout;
9+
10+#[async_test]
11+async fn test_timeline_fills_segmented_history_without_replacing_retained_items() {
12+ let server = MatrixMockServer::new().await;
13+ let client = server.client_builder().build().await;
14+ client.send_queue().set_enabled(false).await;
15+ client.event_cache().subscribe().unwrap();
16+ server.mock_room_state_encryption().plain().mount().await;
17+
18+ let room_id = room_id!("!segmented:example.org");
19+ let room = server.sync_joined_room(&client, room_id).await;
20+ let (cache, _drop_handles) = room.event_cache().await.unwrap();
21+ let event_factory = EventFactory::new().room(room_id).sender(&ALICE);
22+ let old_event_id = event_id!("$old");
23+ let hidden_event_id = event_id!("$hidden");
24+ let new_event_id = event_id!("$new");
25+ cache
26+ .reseed(
27+ vec![event_factory.text_msg("old").event_id(old_event_id).into_event()],
28+ Some("old-cursor".into()),
29+ )
30+ .await
31+ .unwrap();
32+ let gap = cache
33+ .append_history_section(
34+ vec![
35+ event_factory.reaction(old_event_id, "👍").event_id(hidden_event_id).into_event(),
36+ event_factory.text_msg("new").event_id(new_event_id).into_event(),
37+ ],
38+ Some("history-cursor".into()),
39+ )
40+ .await
41+ .unwrap()
42+ .unwrap();
43+
44+ let timeline = room.timeline().await.unwrap();
45+ let (mut timeline_stream_items, mut timeline_stream) = timeline.subscribe().await;
46+ let initial_event_ids: Vec<_> =
47+ timeline_stream_items.iter().filter_map(|item| item.as_event()?.event_id()).collect();
48+ assert_eq!(initial_event_ids, vec![old_event_id, new_event_id]);
49+ let gap_index = timeline_stream_items
50+ .iter()
51+ .position(|item| {
52+ matches!(item.as_virtual(), Some(VirtualTimelineItem::HistoryGap(item_gap)) if item_gap.id == gap.id)
53+ })
54+ .unwrap();
55+ let old_unique_id = timeline_stream_items
56+ .iter()
57+ .find(|item| item.as_event().is_some_and(|event| event.event_id() == Some(old_event_id)))
58+ .unwrap()
59+ .unique_id()
60+ .clone();
61+ let new_unique_id = timeline_stream_items
62+ .iter()
63+ .find(|item| item.as_event().is_some_and(|event| event.event_id() == Some(new_event_id)))
64+ .unwrap()
65+ .unique_id()
66+ .clone();
67+ assert!(gap_index > 0);
68+ assert!(gap_index < timeline_stream_items.len() - 1);
69+
70+ timeline.send(RoomMessageEventContent::text_plain("pending local echo").into()).await.unwrap();
71+ timeout(Duration::from_secs(5), async {
72+ loop {
73+ if timeline_stream_items.iter().any(|item| {
74+ item.as_event().is_some_and(|event| {
75+ event
76+ .content()
77+ .as_message()
78+ .is_some_and(|message| message.body() == "pending local echo")
79+ })
80+ }) {
81+ break;
82+ }
83+ timeline_stream.next().await.unwrap();
84+ timeline_stream_items = timeline.items().await;
85+ }
86+ })
87+ .await
88+ .unwrap();
89+ let pending_unique_id = timeline_stream_items
90+ .iter()
91+ .find(|item| {
92+ item.as_event().is_some_and(|event| {
93+ event
94+ .content()
95+ .as_message()
96+ .is_some_and(|message| message.body() == "pending local echo")
97+ })
98+ })
99+ .unwrap()
100+ .unique_id()
101+ .clone();
102+
103+ server
104+ .mock_room_messages()
105+ .match_from("history-cursor")
106+ .ok(RoomMessagesResponseTemplate::default().end_token("next-cursor").events(vec![
107+ event_factory.text_msg("middle").event_id(event_id!("$middle")),
108+ event_factory.text_msg("old").event_id(old_event_id),
109+ ]))
110+ .mock_once()
111+ .mount()
112+ .await;
113+ cache.fill_history_gap_once(gap.id).await.unwrap();
114+
115+ timeout(Duration::from_secs(5), async {
116+ loop {
117+ timeline_stream_items = timeline.items().await;
118+ let has_middle = timeline_stream_items.iter().any(|item| {
119+ item.as_event().is_some_and(|event| event.event_id() == Some(event_id!("$middle")))
120+ });
121+ let has_gap = timeline_stream_items
122+ .iter()
123+ .any(|item| matches!(item.as_virtual(), Some(VirtualTimelineItem::HistoryGap(_))));
124+ if has_middle && !has_gap {
125+ break;
126+ }
127+ assert!(timeline_stream.next().await.is_some());
128+ }
129+ })
130+ .await
131+ .unwrap();
132+
133+ let final_event_ids: Vec<_> =
134+ timeline_stream_items.iter().filter_map(|item| item.as_event()?.event_id()).collect();
135+ assert_eq!(final_event_ids, vec![old_event_id, event_id!("$middle"), new_event_id]);
136+ let retained_items = [(old_event_id, &old_unique_id), (new_event_id, &new_unique_id)];
137+ for (event_id, unique_id) in retained_items {
138+ let item = timeline_stream_items
139+ .iter()
140+ .find(|item| item.as_event().is_some_and(|event| event.event_id() == Some(event_id)))
141+ .unwrap();
142+ assert_eq!(item.unique_id(), unique_id);
143+ }
144+ let pending_item = timeline_stream_items
145+ .iter()
146+ .find(|item| {
147+ item.as_event().is_some_and(|event| {
148+ event
149+ .content()
150+ .as_message()
151+ .is_some_and(|message| message.body() == "pending local echo")
152+ })
153+ })
154+ .unwrap();
155+ assert_eq!(pending_item.unique_id(), &pending_unique_id);
156+}
crates/matrix-sdk/Cargo.toml+8
@@ -213,5 +213,13 @@ required-features = ["testing", "e2e-encryption"]
213213 name = "history_reseed"
214214 required-features = ["testing"]
215215
216+[[test]]
217+name = "segmented_history"
218+required-features = ["testing"]
219+
220+[[test]]
221+name = "recent_history"
222+required-features = ["testing"]
223+
216224 [lints]
217225 workspace = true
crates/matrix-sdk/src/event_cache/caches/event_focused/mod.rs+9−6
@@ -274,13 +274,15 @@ impl EventFocusedCacheInner {
274274 ) {
275275 // Insert backward gap at the back if we have a token, and the events
276276 // themselves.
277- self.chunk
278- .push_live_events(prev_gap_token.map(|prev_token| Gap { token: prev_token }), &events);
277+ self.chunk.push_live_events(
278+ prev_gap_token.map(|prev_token| Gap { token: prev_token, history: None }),
279+ &events,
280+ );
279281
280282 // Insert forward gap at back if we have a token.
281283 if let Some(next_token) = next_gap_token {
282284 trace!("inserting forward pagination gap at back");
283- self.chunk.push_gap(Gap { token: next_token });
285+ self.chunk.push_gap(Gap { token: next_token, history: None });
284286 }
285287 }
286288
@@ -302,7 +304,7 @@ impl EventFocusedCacheInner {
302304 fn notify_subscribers(&mut self, origin: EventsOrigin) {
303305 let diffs = self.chunk.updates_as_vector_diffs();
304306 if !diffs.is_empty() {
305- let _ = self.sender.send(TimelineVectorDiffs { diffs, origin });
307+ let _ = self.sender.send(TimelineVectorDiffs { diffs, origin, history_gaps: None });
306308 }
307309 }
308310
@@ -365,7 +367,7 @@ impl EventFocusedCacheInner {
365367 events.reverse();
366368
367369 let hit_end = new_token.is_none();
368- let new_gap = new_token.map(|t| Gap { token: t });
370+ let new_gap = new_token.map(|t| Gap { token: t, history: None });
369371
370372 let hide_thread_events = match &self.pagination_mode {
371373 EventFocusedPaginationMode::Room { hide_thread_events } => *hide_thread_events,
@@ -488,7 +490,7 @@ impl EventFocusedCacheInner {
488490 let (events, new_token) = pagination?;
489491
490492 let hit_end = new_token.is_none();
491- let new_gap = new_token.map(|t| Gap { token: t });
493+ let new_gap = new_token.map(|t| Gap { token: t, history: None });
492494
493495 let hide_thread_events = match &self.pagination_mode {
494496 EventFocusedPaginationMode::Room { hide_thread_events } => *hide_thread_events,
@@ -615,6 +617,7 @@ impl EventFocusedCache {
615617 let _ = self.sender.send(TimelineVectorDiffs {
616618 diffs: vec![VectorDiff::Clear],
617619 origin: EventsOrigin::Cache,
620+ history_gaps: None,
618621 });
619622 let _ = self.linked_chunk_update_sender.send(RoomEventCacheLinkedChunkUpdate {
620623 updates: vec![Update::Clear],
crates/matrix-sdk/src/event_cache/caches/event_linked_chunk.rs+28−7
@@ -318,6 +318,27 @@ impl EventLinkedChunk {
318318 self.chunks.push_items_back(events.iter().cloned());
319319 }
320320
321+ pub(super) fn insert_history_before(
322+ &mut self,
323+ before: Option<&ruma::EventId>,
324+ events: Vec<Event>,
325+ gap: Option<Gap>,
326+ ) {
327+ let position = before.and_then(|id| {
328+ self.events().find_map(|(position, event)| {
329+ (event.event_id().as_deref() == Some(id)).then_some(position)
330+ })
331+ });
332+ if let Some(position) = position {
333+ self.chunks.insert_items_at(position, events).expect("existing event position");
334+ if let Some(gap) = gap {
335+ self.chunks.insert_gap_at(gap, position).expect("existing event position");
336+ }
337+ } else {
338+ self.push_live_events(gap, &events);
339+ }
340+ }
341+
321342 /// Add events from a backwards pagination for this linked chunk by updating
322343 /// the in-memory linked chunk with the results.
323344 ///
@@ -598,7 +619,7 @@ fn chunk_debug_string(
598619 order_tracker: &OrderTracker<Event, Gap>,
599620 ) -> String {
600621 match content {
601- ChunkContent::Gap(Gap { token: prev_token }) => {
622+ ChunkContent::Gap(Gap { token: prev_token, .. }) => {
602623 format!("gap['{prev_token}']")
603624 }
604625 ChunkContent::Items(vec) => {
@@ -695,7 +716,7 @@ mod tests {
695716 let mut linked_chunk = EventLinkedChunk::new();
696717
697718 linked_chunk.chunks.push_items_back([event_0]);
698- linked_chunk.chunks.push_gap_back(Gap { token: "hello".to_owned() });
719+ linked_chunk.chunks.push_gap_back(Gap { token: "hello".to_owned(), history: None });
699720
700721 let gap_chunk_id = linked_chunk
701722 .chunks()
@@ -735,9 +756,9 @@ mod tests {
735756 let mut linked_chunk = EventLinkedChunk::new();
736757
737758 linked_chunk.chunks.push_items_back([event_0, event_1]);
738- linked_chunk.chunks.push_gap_back(Gap { token: "middle".to_owned() });
759+ linked_chunk.chunks.push_gap_back(Gap { token: "middle".to_owned(), history: None });
739760 linked_chunk.chunks.push_items_back([event_2]);
740- linked_chunk.chunks.push_gap_back(Gap { token: "end".to_owned() });
761+ linked_chunk.chunks.push_gap_back(Gap { token: "end".to_owned(), history: None });
741762
742763 // Remove the first gap.
743764 let first_gap_id = linked_chunk
@@ -770,7 +791,7 @@ mod tests {
770791 // Push some events.
771792 let mut linked_chunk = EventLinkedChunk::new();
772793 linked_chunk.chunks.push_items_back([event_0, event_1]);
773- linked_chunk.chunks.push_gap_back(Gap { token: "hello".to_owned() });
794+ linked_chunk.chunks.push_gap_back(Gap { token: "hello".to_owned(), history: None });
774795 linked_chunk.chunks.push_items_back([event_2, event_3]);
775796
776797 assert_events_eq!(
@@ -843,7 +864,7 @@ mod tests {
843864 // Push some events.
844865 let mut linked_chunk = EventLinkedChunk::new();
845866 linked_chunk.chunks.push_items_back([event_0, event_1]);
846- linked_chunk.chunks.push_gap_back(Gap { token: "raclette".to_owned() });
867+ linked_chunk.chunks.push_gap_back(Gap { token: "raclette".to_owned(), history: None });
847868 linked_chunk.chunks.push_items_back([event_2]);
848869
849870 // Read the updates as `VectorDiff`.
@@ -898,7 +919,7 @@ mod tests {
898919 .into_event(),
899920 event_factory.text_msg("you").event_id(event_id!("$2")).into_event(),
900921 ]);
901- linked_chunk.chunks.push_gap_back(Gap { token: "raclette".to_owned() });
922+ linked_chunk.chunks.push_gap_back(Gap { token: "raclette".to_owned(), history: None });
902923
903924 // Flush updates to the order tracker.
904925 let _ = linked_chunk.updates_as_vector_diffs();
crates/matrix-sdk/src/event_cache/caches/mod.rs+3
@@ -188,6 +188,7 @@ impl<'c> ResetCaches<'c> {
188188 let updates_as_vector_diffs = room_state.reset().await?;
189189 room.update_sender().send(
190190 room::RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
191+ history_gaps: Some(Vec::new()),
191192 diffs: updates_as_vector_diffs,
192193 origin: EventsOrigin::Cache,
193194 }),
@@ -202,6 +203,8 @@ impl<'c> ResetCaches<'c> {
202203 /// A diff update for an event cache timeline represented as a vector.
203204 #[derive(Clone, Debug)]
204205 pub struct TimelineVectorDiffs {
206+ #[doc = "The ordered room gaps after these diffs, or no change for focused caches."]
207+ pub history_gaps: Option<Vec<room::HistoryGapPosition>>,
205208 /// New vector diff for the thread timeline.
206209 pub diffs: Vec<VectorDiff<Event>>,
207210 /// The origin that triggered this update.
crates/matrix-sdk/src/event_cache/caches/pinned_events/mod.rs+2−1
@@ -189,7 +189,8 @@ impl<'a> PinnedEventCacheStateLockWriteGuard<'a> {
189189 fn notify_subscribers(&mut self, origin: EventsOrigin) {
190190 let diffs = self.state.chunk.updates_as_vector_diffs();
191191 if !diffs.is_empty() {
192- let _ = self.state.sender.send(TimelineVectorDiffs { diffs, origin });
192+ let _ =
193+ self.state.sender.send(TimelineVectorDiffs { diffs, origin, history_gaps: None });
193194 }
194195 }
195196 }
crates/matrix-sdk/src/event_cache/caches/room/append_history_section.md+16
@@ -0,0 +1,16 @@
1+Install independently fetched events in topological order, oldest first, while
2+preserving retained history and its pagination cursors. Duplicate room event IDs
3+are retained once. A wholly duplicated or empty batch creates no new section.
4+
5+When retained events exist, the returned gap connects their newest event to the
6+oldest newly inserted event. Its cursor belongs to the supplied newer section.
7+An absent cursor leaves an unavailable interval rather than claiming continuity.
8+Without retained events, the cursor remains ordinary oldest-history pagination.
9+
10+Events and gap metadata commit in one encrypted event cache transaction before
11+publication. A failed write preserves the previous cache. Installation continues
12+if the caller drops the future. Active pinned, thread, and focused caches are not
13+cleared. Completion means cache publication, not application mirror consumption.
14+
15+This low-level operation does not reconcile the freshness of an independently
16+fetched batch with live arrivals. Callers must coordinate that acquisition.
crates/matrix-sdk/src/event_cache/caches/room/fill_history_gap_once.md+20
@@ -0,0 +1,20 @@
1+Request one backward page of up to 50 server events using this gap's saved cursor.
2+Ordinary oldest-history pagination and other internal gaps keep their cursors.
3+
4+Only reaching the recorded older boundary verifies a join. Short or empty pages
5+with advancing cursors leave the gap idle. A repeated cursor pauses it. Exhausting
6+accessible history leaves it unavailable and preserves retained events. Joined
7+identities remain persisted so repeated calls can return their terminal outcome.
8+
9+Dropping the future before the response cancels the request without advancing
10+the cursor. Once installation starts, the page and cursor commit together even
11+if the caller stops waiting. Request failures retain the last successful cursor.
12+The original SDK error is returned after its failure state has been persisted.
13+
14+The room coordinator permits one active targeted fill or section installation.
15+Concurrent fills return `Busy`. Recent load callers share an in-flight operation.
16+Dropping one recent caller leaves the operation active for the remaining callers.
17+Dropping all callers cancels fetching, while a started commit runs to completion.
18+The coordinator keeps no strong cache reference while idle. Cache publication
19+precedes completion, while application mirror publication remains the caller's
20+responsibility. `wait_for_history` waits for earlier operations and their commits.
crates/matrix-sdk/src/event_cache/caches/room/history.rs+551
@@ -0,0 +1,551 @@
1+use std::{
2+ collections::{HashSet, VecDeque},
3+ sync::Weak,
4+};
5+
6+pub use matrix_sdk_base::event_cache::{HistoryGap, HistoryGapStatus};
7+use matrix_sdk_base::{
8+ event_cache::{Event, Gap},
9+ linked_chunk::ChunkContent,
10+};
11+use ruma::{OwnedEventId, OwnedTransactionId, TransactionId};
12+use tokio::sync::{mpsc, oneshot, watch};
13+
14+use super::{RoomEventCache, RoomEventCacheInner};
15+use crate::{
16+ event_cache::{EventCacheError, Result, caches::event_linked_chunk::EventLinkedChunk},
17+ room::MessagesOptions,
18+};
19+
20+#[derive(Clone, Debug, PartialEq, Eq)]
21+#[doc = "A durable interval positioned before the next identified raw event in cache order."]
22+pub struct HistoryGapPosition {
23+ #[doc = "The durable identity and continuity state of the interval."]
24+ pub gap: HistoryGap,
25+ #[doc = "Absent when no identified event follows the gap in the loaded window."]
26+ pub next_event_id: Option<OwnedEventId>,
27+}
28+
29+pub(super) fn gap_positions(candidate: &EventLinkedChunk) -> Vec<HistoryGapPosition> {
30+ let mut positions: Vec<HistoryGapPosition> = Vec::new();
31+ let mut pending = 0;
32+ for chunk in candidate.chunks() {
33+ match chunk.content() {
34+ ChunkContent::Gap(gap) => {
35+ if let Some(gap) = &gap.history {
36+ positions.push(HistoryGapPosition { gap: gap.clone(), next_event_id: None });
37+ }
38+ }
39+ ChunkContent::Items(events) => {
40+ if let Some(id) = events.iter().find_map(Event::event_id) {
41+ for position in &mut positions[pending..] {
42+ position.next_event_id = Some(id.clone());
43+ }
44+ pending = positions.len();
45+ }
46+ }
47+ }
48+ }
49+ positions
50+}
51+
52+#[derive(Clone, Debug, PartialEq, Eq)]
53+#[doc = "A bounded gap fill never falls back to another gap when its target disappears."]
54+pub enum HistoryGapFillOutcome {
55+ #[doc = "Also returned without a network request for a joined or unavailable interval."]
56+ Updated {
57+ #[doc = "The durable outcome, including its original boundary identities."]
58+ gap: HistoryGap,
59+ #[doc = "Newly inserted events after room-wide deduplication."]
60+ events: usize,
61+ },
62+ #[doc = "The identity is unknown or was removed by an explicit cache replacement."]
63+ NotFound,
64+ #[doc = "Another history operation owns the room coordinator. No request was started."]
65+ Busy,
66+}
67+
68+#[derive(Clone, Debug)]
69+#[doc = "A fresh server window bounded by four requests of fifty raw events."]
70+pub struct RecentHistoryOutcome {
71+ #[doc = "Number of server pages consumed, including empty pages."]
72+ pub pages: u32,
73+ #[doc = "Number of raw events returned by the server."]
74+ pub events: usize,
75+ #[doc = "The durable interval, absent when no retained boundary needs joining."]
76+ pub gap: Option<HistoryGap>,
77+}
78+
79+pub(super) struct HistoryRequest {
80+ operation: HistoryOperation,
81+ reply: oneshot::Sender<Result<HistoryResponse>>,
82+}
83+
84+enum HistoryOperation {
85+ Recent,
86+ Fill(OwnedTransactionId),
87+ Append(Vec<Event>, Option<String>),
88+ Restore,
89+ Barrier,
90+}
91+
92+#[derive(Clone)]
93+enum HistoryResponse {
94+ Recent(RecentHistoryOutcome),
95+ Fill(HistoryGapFillOutcome),
96+ Append(Option<HistoryGap>),
97+ Restored,
98+}
99+
100+async fn coordinate_history(
101+ weak_cache: Weak<RoomEventCacheInner>,
102+ mut requests: mpsc::Receiver<HistoryRequest>,
103+) {
104+ let mut pending = VecDeque::new();
105+ loop {
106+ let request = if let Some(request) = pending.pop_front() {
107+ request
108+ } else if let Some(request) = requests.recv().await {
109+ request
110+ } else {
111+ return;
112+ };
113+ let HistoryRequest { operation, reply } = request;
114+ if reply.is_closed() {
115+ continue;
116+ }
117+ let Some(inner) = weak_cache.upgrade() else {
118+ return;
119+ };
120+ let cache = RoomEventCache { inner };
121+ let recent = matches!(operation, HistoryOperation::Recent);
122+ let mut waiters = vec![reply];
123+ let (cancel, cancellation) = watch::channel(false);
124+ let operation = async {
125+ match operation {
126+ HistoryOperation::Recent => {
127+ cache.load_recent_history_inner(cancellation).await.map(HistoryResponse::Recent)
128+ }
129+ HistoryOperation::Fill(id) => {
130+ cache.fill_history_gap_inner(id, cancellation).await.map(HistoryResponse::Fill)
131+ }
132+ HistoryOperation::Append(events, cursor) => cache
133+ .append_history_section_inner(events, cursor)
134+ .await
135+ .map(HistoryResponse::Append),
136+ HistoryOperation::Restore => {
137+ cache.load_retained_history_inner().await.map(|()| HistoryResponse::Restored)
138+ }
139+ HistoryOperation::Barrier => Ok(HistoryResponse::Restored),
140+ }
141+ };
142+ tokio::pin!(operation);
143+ let mut cancelled = false;
144+ let mut accepting = true;
145+ loop {
146+ enum Action {
147+ Complete(Result<HistoryResponse>),
148+ Request(Option<HistoryRequest>),
149+ Cancel,
150+ }
151+ let action = tokio::select! {
152+ biased;
153+ result = &mut operation => Action::Complete(result),
154+ _ = futures_util::future::join_all(waiters.iter_mut().map(|reply| reply.closed())), if !cancelled => Action::Cancel,
155+ request = requests.recv(), if accepting => Action::Request(request),
156+ };
157+ match action {
158+ Action::Complete(result) => {
159+ for reply in waiters {
160+ let _ = reply.send(result.clone());
161+ }
162+ break;
163+ }
164+ Action::Request(Some(request)) => {
165+ if recent && !cancelled && matches!(request.operation, HistoryOperation::Recent)
166+ {
167+ waiters.push(request.reply);
168+ } else if !cancelled && matches!(request.operation, HistoryOperation::Fill(_)) {
169+ let _ = request
170+ .reply
171+ .send(Ok(HistoryResponse::Fill(HistoryGapFillOutcome::Busy)));
172+ } else {
173+ pending.push_back(request);
174+ }
175+ }
176+ Action::Request(None) => accepting = false,
177+ Action::Cancel => {
178+ cancelled = true;
179+ let _ = cancel.send(true);
180+ }
181+ }
182+ }
183+ }
184+}
185+
186+impl RoomEventCache {
187+ async fn request_history(&self, operation: HistoryOperation) -> Result<HistoryResponse> {
188+ let sender = self.inner.history_coordinator.get_or_init(|| {
189+ let (sender, requests) = mpsc::channel(16);
190+ matrix_sdk_common::executor::spawn(coordinate_history(
191+ std::sync::Arc::downgrade(&self.inner),
192+ requests,
193+ ));
194+ sender
195+ });
196+ let (reply, response) = oneshot::channel();
197+ sender
198+ .send(HistoryRequest { operation, reply })
199+ .await
200+ .map_err(|_| EventCacheError::ClientDropped)?;
201+ response.await.map_err(|_| EventCacheError::ClientDropped)?
202+ }
203+
204+ #[doc = "Waits for earlier history operations, including commits whose callers stopped waiting."]
205+ pub async fn wait_for_history(&self) -> Result<()> {
206+ self.request_history(HistoryOperation::Barrier).await.map(|_| ())
207+ }
208+
209+ #[doc = "Fetches from a fresh server position and retains sync arrivals during acquisition."]
210+ pub async fn load_recent_history(&self) -> Result<RecentHistoryOutcome> {
211+ match self.request_history(HistoryOperation::Recent).await? {
212+ HistoryResponse::Recent(outcome) => Ok(outcome),
213+ _ => unreachable!("coordinator response matches request"),
214+ }
215+ }
216+
217+ async fn load_recent_history_inner(
218+ &self,
219+ mut cancellation: watch::Receiver<bool>,
220+ ) -> Result<RecentHistoryOutcome> {
221+ let anchor = {
222+ let state = self.inner.state.write().await?;
223+ state.history_candidate().await?.revents().find_map(|(_, event)| event.event_id())
224+ };
225+ let room = self.inner.weak_room.get().ok_or(EventCacheError::ClientDropped)?;
226+ let mut cursor = None;
227+ let mut visited = Vec::new();
228+ let mut events = Vec::new();
229+ let mut pages = 0;
230+ let mut raw_count = 0;
231+ let mut joined = false;
232+ let mut stalled = false;
233+ for _ in 0..4 {
234+ let mut options = MessagesOptions::backward();
235+ options.from = cursor.clone();
236+ options.limit = 50u16.into();
237+ if *cancellation.borrow() {
238+ return Err(EventCacheError::ClientDropped);
239+ }
240+ let response = tokio::select! {
241+ biased;
242+ _ = cancellation.changed() => return Err(EventCacheError::ClientDropped),
243+ response = room.messages(options) => response,
244+ }
245+ .map_err(|error| EventCacheError::PaginationError(std::sync::Arc::new(error)))?;
246+ pages += 1;
247+ raw_count += response.chunk.len();
248+ let mut page = response.chunk;
249+ if let Some(index) = anchor.as_ref().and_then(|anchor| {
250+ page.iter().position(|event| event.event_id().as_ref() == Some(anchor))
251+ }) {
252+ page.truncate(index);
253+ joined = true;
254+ }
255+ events.extend(page);
256+ if let Some(consumed) = cursor.take() {
257+ visited.push(consumed);
258+ }
259+ cursor = response.end;
260+ stalled = cursor.as_ref().is_some_and(|next| visited.contains(next));
261+ if joined || cursor.is_none() || stalled {
262+ break;
263+ }
264+ }
265+ events.reverse();
266+ {
267+ let mut state = self.inner.state.write().await?;
268+ let mut candidate = state.history_candidate().await?;
269+ let baseline_count = candidate.events().count();
270+ let fallback = {
271+ let mut after_anchor = anchor.is_none();
272+ candidate.events().find_map(|(_, event)| {
273+ let id = event.event_id();
274+ if after_anchor {
275+ return id;
276+ }
277+ if id == anchor {
278+ after_anchor = true;
279+ }
280+ None
281+ })
282+ };
283+ let newer_boundary = events.iter().find_map(Event::event_id).or(fallback.clone());
284+ let history = anchor.map(|anchor| HistoryGap {
285+ id: TransactionId::new(),
286+ older_boundary: Some(anchor),
287+ newer_boundary: newer_boundary.clone(),
288+ status: if joined {
289+ HistoryGapStatus::Joined
290+ } else if cursor.is_none() {
291+ HistoryGapStatus::Unavailable
292+ } else if stalled {
293+ HistoryGapStatus::Paused
294+ } else {
295+ HistoryGapStatus::Idle
296+ },
297+ visited_cursors: visited,
298+ });
299+ let mut before = fallback;
300+ let mut inserted = Vec::new();
301+ for event in events.into_iter().rev() {
302+ let id = event.event_id();
303+ if id.as_ref().is_none_or(|id| {
304+ !candidate.events().any(|(_, event)| event.event_id().as_ref() == Some(id))
305+ }) {
306+ candidate.insert_history_before(before.as_deref(), vec![event.clone()], None);
307+ inserted.push(event);
308+ }
309+ if id.is_some() {
310+ before = id;
311+ }
312+ }
313+ if history.is_some() || cursor.is_some() {
314+ candidate.insert_history_before(
315+ before.as_deref(),
316+ Vec::new(),
317+ Some(Gap { token: cursor.unwrap_or_default(), history: history.clone() }),
318+ );
319+ }
320+ inserted.reverse();
321+ state.commit_history_candidate(candidate, inserted, baseline_count).await?;
322+ Ok(RecentHistoryOutcome { pages, events: raw_count, gap: history })
323+ }
324+ }
325+
326+ #[doc = "Loads retained sections before attaching a live timeline subscriber."]
327+ pub async fn load_retained_history(&self) -> Result<()> {
328+ match self.request_history(HistoryOperation::Restore).await? {
329+ HistoryResponse::Restored => Ok(()),
330+ _ => unreachable!("coordinator response matches request"),
331+ }
332+ }
333+
334+ async fn load_retained_history_inner(&self) -> Result<()> {
335+ let mut state = self.inner.state.write().await?;
336+ let candidate = state.history_candidate().await?;
337+ if !gaps(&candidate).is_empty()
338+ && candidate.chunks().count() != state.room_linked_chunk().chunks().count()
339+ {
340+ let baseline_count = candidate.events().count();
341+ state.commit_history_candidate(candidate, Vec::new(), baseline_count).await?;
342+ }
343+ Ok(())
344+ }
345+
346+ #[doc = "Returns loaded events and their gaps under the same cache lock."]
347+ pub async fn history_snapshot(&self) -> Result<(Vec<Event>, Vec<HistoryGapPosition>)> {
348+ let state = self.inner.state.read().await?;
349+ let chunk = state.room_linked_chunk();
350+ Ok((chunk.events().map(|(_, event)| event.clone()).collect(), gap_positions(chunk)))
351+ }
352+
353+ #[doc = "Includes terminal identities and gaps outside the currently loaded cache window."]
354+ pub async fn history_gaps(&self) -> Result<Vec<HistoryGap>> {
355+ let state = self.inner.state.write().await?;
356+ Ok(gaps(&state.history_candidate().await?))
357+ }
358+
359+ #[doc = include_str!("append_history_section.md")]
360+ pub async fn append_history_section(
361+ &self,
362+ events: Vec<Event>,
363+ prev_batch: Option<String>,
364+ ) -> Result<Option<HistoryGap>> {
365+ match self.request_history(HistoryOperation::Append(events, prev_batch)).await? {
366+ HistoryResponse::Append(gap) => Ok(gap),
367+ _ => unreachable!("coordinator response matches request"),
368+ }
369+ }
370+
371+ async fn append_history_section_inner(
372+ &self,
373+ events: Vec<Event>,
374+ prev_batch: Option<String>,
375+ ) -> Result<Option<HistoryGap>> {
376+ {
377+ let mut state = self.inner.state.write().await?;
378+ let mut candidate = state.history_candidate().await?;
379+ let baseline_count = candidate.events().count();
380+ let older_boundary = candidate.revents().find_map(|(_, event)| event.event_id());
381+ let mut known: HashSet<_> =
382+ candidate.events().filter_map(|(_, event)| event.event_id()).collect();
383+ let events: Vec<_> = events
384+ .into_iter()
385+ .filter(|event| event.event_id().is_none_or(|id| known.insert(id)))
386+ .collect();
387+ if events.is_empty() {
388+ return Ok(None);
389+ }
390+ let history = older_boundary.map(|older_boundary| HistoryGap {
391+ id: TransactionId::new(),
392+ older_boundary: Some(older_boundary),
393+ newer_boundary: events.iter().find_map(Event::event_id),
394+ visited_cursors: Vec::new(),
395+ status: if prev_batch.is_some() {
396+ HistoryGapStatus::Idle
397+ } else {
398+ HistoryGapStatus::Unavailable
399+ },
400+ });
401+ let gap = if history.is_some() || prev_batch.is_some() {
402+ Some(Gap { token: prev_batch.unwrap_or_default(), history: history.clone() })
403+ } else {
404+ None
405+ };
406+ candidate.push_live_events(gap, &events);
407+ state.commit_history_candidate(candidate, events, baseline_count).await?;
408+ Ok(history)
409+ }
410+ }
411+
412+ #[doc = include_str!("fill_history_gap_once.md")]
413+ pub async fn fill_history_gap_once(
414+ &self,
415+ gap_id: OwnedTransactionId,
416+ ) -> Result<HistoryGapFillOutcome> {
417+ match self.request_history(HistoryOperation::Fill(gap_id)).await? {
418+ HistoryResponse::Fill(outcome) => Ok(outcome),
419+ _ => unreachable!("coordinator response matches request"),
420+ }
421+ }
422+
423+ async fn fill_history_gap_inner(
424+ &self,
425+ gap_id: OwnedTransactionId,
426+ mut cancellation: watch::Receiver<bool>,
427+ ) -> Result<HistoryGapFillOutcome> {
428+ let initial = {
429+ let state = self.inner.state.write().await?;
430+ find_gap(&state.history_candidate().await?, &gap_id)
431+ };
432+ let Some(initial) = initial else {
433+ return Ok(HistoryGapFillOutcome::NotFound);
434+ };
435+ let history = initial.history.as_ref().expect("find_gap only returns history gaps");
436+ if matches!(history.status, HistoryGapStatus::Joined | HistoryGapStatus::Unavailable) {
437+ return Ok(HistoryGapFillOutcome::Updated { gap: history.clone(), events: 0 });
438+ }
439+ let room = self.inner.weak_room.get().ok_or(EventCacheError::ClientDropped)?;
440+ let mut options = MessagesOptions::backward().from(initial.token.as_str());
441+ options.limit = 50u16.into();
442+ if *cancellation.borrow() {
443+ return Err(EventCacheError::ClientDropped);
444+ }
445+ let response = tokio::select! {
446+ biased;
447+ _ = cancellation.changed() => return Err(EventCacheError::ClientDropped),
448+ response = room.messages(options) => response,
449+ };
450+ {
451+ let mut state = self.inner.state.write().await?;
452+ let mut candidate = state.history_candidate().await?;
453+ let baseline_count = candidate.events().count();
454+ let Some(current) = find_gap(&candidate, &gap_id) else {
455+ return Ok(HistoryGapFillOutcome::NotFound);
456+ };
457+ if current.token != initial.token {
458+ return Ok(HistoryGapFillOutcome::Busy);
459+ }
460+ if let Some(gap) = current.history.as_ref()
461+ && matches!(gap.status, HistoryGapStatus::Joined | HistoryGapStatus::Unavailable)
462+ {
463+ return Ok(HistoryGapFillOutcome::Updated { gap: gap.clone(), events: 0 });
464+ }
465+ let (events, next, failure) = match response {
466+ Ok(response) => (response.chunk, response.end, None),
467+ Err(error) => (Vec::new(), Some(current.token.clone()), Some(error)),
468+ };
469+ let (gap, events) = apply_page(
470+ &mut candidate,
471+ &gap_id,
472+ events,
473+ next,
474+ failure.as_ref().map(ToString::to_string),
475+ );
476+ let count = events.len();
477+ state.commit_history_candidate(candidate, events, baseline_count).await?;
478+ if let Some(error) = failure {
479+ return Err(EventCacheError::PaginationError(std::sync::Arc::new(error)));
480+ }
481+ Ok(HistoryGapFillOutcome::Updated { gap, events: count })
482+ }
483+ }
484+}
485+
486+fn gaps(candidate: &EventLinkedChunk) -> Vec<HistoryGap> {
487+ candidate
488+ .chunks()
489+ .filter_map(|chunk| match chunk.content() {
490+ ChunkContent::Gap(gap) => gap.history.clone(),
491+ _ => None,
492+ })
493+ .collect()
494+}
495+
496+fn find_gap(candidate: &EventLinkedChunk, gap_id: &TransactionId) -> Option<Gap> {
497+ candidate.chunks().find_map(|chunk| match chunk.content() {
498+ ChunkContent::Gap(gap)
499+ if gap.history.as_ref().is_some_and(|history| history.id == gap_id) =>
500+ {
501+ Some(gap.clone())
502+ }
503+ _ => None,
504+ })
505+}
506+
507+fn apply_page(
508+ candidate: &mut EventLinkedChunk,
509+ gap_id: &TransactionId,
510+ mut events: Vec<Event>,
511+ next: Option<String>,
512+ failure: Option<String>,
513+) -> (HistoryGap, Vec<Event>) {
514+ let gap_chunk = candidate.chunks().find(|chunk| matches!(
515+ chunk.content(), ChunkContent::Gap(gap) if gap.history.as_ref().is_some_and(|history| history.id == gap_id)
516+ )).expect("the gap was checked under the state lock");
517+ let chunk_id = gap_chunk.identifier();
518+ let ChunkContent::Gap(current) = gap_chunk.content() else { unreachable!() };
519+ let mut gap = current.clone();
520+ let history = gap.history.as_mut().expect("the gap was checked under the state lock");
521+ let overlap = history.older_boundary.as_ref().and_then(|boundary| {
522+ events.iter().position(|event| event.event_id().as_ref() == Some(boundary))
523+ });
524+ history.status = if let Some(message) = failure {
525+ HistoryGapStatus::Failed { message }
526+ } else if let Some(index) = overlap {
527+ events.truncate(index);
528+ HistoryGapStatus::Joined
529+ } else if next.is_none() {
530+ HistoryGapStatus::Unavailable
531+ } else if next.as_ref() == Some(&gap.token)
532+ || next.as_ref().is_some_and(|next| history.visited_cursors.contains(next))
533+ {
534+ HistoryGapStatus::Paused
535+ } else {
536+ HistoryGapStatus::Idle
537+ };
538+ if !history.visited_cursors.contains(&gap.token) {
539+ history.visited_cursors.push(gap.token.clone());
540+ }
541+ if let Some(next) = next {
542+ gap.token = next;
543+ }
544+ let mut known: HashSet<_> =
545+ candidate.events().filter_map(|(_, event)| event.event_id()).collect();
546+ events.retain(|event| event.event_id().is_none_or(|id| known.insert(id)));
547+ events.reverse();
548+ let history = gap.history.as_ref().expect("history metadata was preserved").clone();
549+ candidate.push_backwards_pagination_events(Some(chunk_id), Some(gap), &events);
550+ (history, events)
551+}
crates/matrix-sdk/src/event_cache/caches/room/mod.rs+91−36
@@ -12,7 +12,9 @@
1212 // See the License for the specific language governing permissions and
1313 // limitations under the License.
1414
15+mod history;
1516 pub mod pagination;
17+pub use history::{HistoryGapFillOutcome, HistoryGapPosition, RecentHistoryOutcome};
1618 mod state;
1719 mod subscriber;
1820 mod updates;
@@ -121,6 +123,14 @@ impl RoomEventCache {
121123 /// subscriber. Creating, and especially dropping, a
122124 /// [`RoomEventCacheSubscriber`] isn't free, as it triggers side-effects.
123125 pub async fn subscribe(&self) -> Result<(Vec<Event>, RoomEventCacheSubscriber)> {
126+ let (events, _, subscriber) = self.subscribe_with_history().await?;
127+ Ok((events, subscriber))
128+ }
129+
130+ #[doc = "Subscribes after taking an atomic snapshot of loaded events and ordered gaps."]
131+ pub async fn subscribe_with_history(
132+ &self,
133+ ) -> Result<(Vec<Event>, Vec<HistoryGapPosition>, RoomEventCacheSubscriber)> {
124134 let state = self.inner.state.read().await?;
125135 let events =
126136 state.room_linked_chunk().events().map(|(_position, item)| item.clone()).collect();
@@ -136,7 +146,7 @@ impl RoomEventCache {
136146 subscriber_count.clone(),
137147 );
138148
139- Ok((events, subscriber))
149+ Ok((events, state.history_gap_positions(), subscriber))
140150 }
141151
142152 /// Subscribe to thread for a given root event, and get a (maybe empty)
@@ -363,6 +373,7 @@ impl RoomEventCache {
363373 self.inner.update_sender.send(
364374 RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
365375 diffs: updates_as_vector_diffs,
376+ history_gaps: Some(history::gap_positions(state.room_linked_chunk())),
366377 origin: EventsOrigin::Cache,
367378 }),
368379 Some(RoomEventCacheGenericUpdate { room_id: self.inner.room_id.clone() }),
@@ -392,6 +403,7 @@ impl RoomEventCache {
392403 cache.inner.update_sender.send(
393404 RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
394405 diffs: vec![eyeball_im::VectorDiff::Reset { values: items }],
406+ history_gaps: Some(history::gap_positions(state.room_linked_chunk())),
395407 origin: EventsOrigin::Cache,
396408 }),
397409 Some(RoomEventCacheGenericUpdate { room_id: cache.inner.room_id.clone() }),
@@ -478,6 +490,7 @@ impl RoomEventCache {
478490
479491 /// The (non-cloneable) details of the `RoomEventCache`.
480492 pub(super) struct RoomEventCacheInner {
493+ history_coordinator: std::sync::OnceLock<mpsc::Sender<history::HistoryRequest>>,
481494 pub generation: Arc<AtomicU64>,
482495 /// The room id for this room.
483496 room_id: OwnedRoomId,
@@ -516,6 +529,7 @@ impl RoomEventCacheInner {
516529 ) -> Self {
517530 Self {
518531 generation,
532+ history_coordinator: Default::default(),
519533 room_id,
520534 weak_room,
521535 state,
@@ -632,8 +646,6 @@ impl RoomEventCacheInner {
632646 let (stored_prev_batch_token, timeline_event_diffs) =
633647 state.handle_sync(timeline, &ephemeral_events).await?;
634648
635- drop(state);
636-
637649 // Now that all events have been added, we can trigger the
638650 // `pagination_token_notifier`.
639651 if stored_prev_batch_token {
@@ -646,12 +658,15 @@ impl RoomEventCacheInner {
646658 self.update_sender.send(
647659 RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
648660 diffs: timeline_event_diffs,
661+ history_gaps: Some(history::gap_positions(state.room_linked_chunk())),
649662 origin: EventsOrigin::Sync,
650663 }),
651664 Some(RoomEventCacheGenericUpdate { room_id: self.room_id.clone() }),
652665 );
653666 }
654667
668+ drop(state);
669+
655670 if !ephemeral_events.is_empty() {
656671 self.update_sender
657672 .send(RoomEventCacheUpdate::AddEphemeralEvents { events: ephemeral_events }, None);
@@ -1188,7 +1203,7 @@ mod timed_tests {
11881203 // Chunk IDs aren't supposed to be ordered, so use a random value here.
11891204 new: ChunkIdentifier::new(42),
11901205 next: None,
1191- gap: Gap { token: "comté".to_owned() },
1206+ gap: Gap { token: "comté".to_owned(), history: None },
11921207 },
11931208 // Another items chunk, non-empty this time.
11941209 Update::NewItemsChunk {
@@ -1345,7 +1360,7 @@ mod timed_tests {
13451360 // Chunk IDs aren't supposed to be ordered, so use a random value here.
13461361 new: ChunkIdentifier::new(42),
13471362 next: None,
1348- gap: Gap { token: "cheddar".to_owned() },
1363+ gap: Gap { token: "cheddar".to_owned(), history: None },
13491364 },
13501365 // Another items chunk, non-empty this time.
13511366 Update::NewItemsChunk {
@@ -1893,10 +1908,8 @@ mod timed_tests {
18931908 }
18941909 }
18951910
1896- // Handle a gappy sync with two events (including one duplicate, so
1897- // deduplication kicks in), so that the linked chunk is shrunk to the
1898- // last chunk, and that the linked chunk only contains the last two
1899- // events.
1911+ // Handle a limited sync with an exact overlap at the retained newest
1912+ // event. The overlap proves continuity with the cached timeline.
19001913 let evid4 = event_id!("$4");
19011914 room_event_cache
19021915 .handle_joined_room_update(JoinedRoomUpdate {
@@ -1914,37 +1927,79 @@ mod timed_tests {
19141927 let state = room_event_cache.inner.state.read().await.unwrap();
19151928 let room_linked_chunk = state.room_linked_chunk();
19161929
1917- // After the shrink, only evid3 and evid4 are loaded.
1918- let mut events = room_linked_chunk.events();
1919-
1920- let (pos, ev) = events.next().unwrap();
1921- assert_eq!(ev.event_id().as_deref(), Some(evid3));
1922- assert_eq!(room_linked_chunk.event_order(pos), Some(2));
1930+ let event_ids: Vec<_> =
1931+ room_linked_chunk.events().filter_map(|(_, event)| event.event_id()).collect();
1932+ assert_eq!(event_ids, vec![evid1, evid2, evid3, evid4]);
1933+ }
19231934
1924- let (pos, ev) = events.next().unwrap();
1925- assert_eq!(ev.event_id().as_deref(), Some(evid4));
1926- assert_eq!(room_linked_chunk.event_order(pos), Some(3));
1935+ assert!(room_event_cache.history_gaps().await.unwrap().is_empty());
1936+ }
19271937
1928- // No other loaded events.
1929- assert!(events.next().is_none());
1938+ #[async_test]
1939+ async fn test_limited_sync_discontinuities_survive_reload() {
1940+ let room_id = room_id!("!limited-history:saucisse.bzh");
1941+ let client = MockClientBuilder::new(None).build().await;
1942+ let event_cache = client.event_cache();
1943+ event_cache.subscribe().unwrap();
1944+ client.base_client().get_or_create_room(room_id, RoomState::Joined);
1945+ let room = client.get_room(room_id).unwrap();
1946+ let (room_event_cache, _drop_handles) = room.event_cache().await.unwrap();
1947+ let f = EventFactory::new().room(room_id).sender(*ALICE);
1948+ let old_id = event_id!("$limited-old");
1949+ let cursor_gap_id = event_id!("$limited-cursor-gap");
1950+ let cursorless_gap_id = event_id!("$limited-cursorless-gap");
19301951
1931- // But we can still get the order of previous events.
1932- assert_eq!(
1933- room_linked_chunk.event_order(Position::new(ChunkIdentifier::new(0), 0)),
1934- Some(0)
1935- );
1936- assert_eq!(
1937- room_linked_chunk.event_order(Position::new(ChunkIdentifier::new(0), 1)),
1938- Some(1)
1939- );
1952+ room_event_cache
1953+ .handle_joined_room_update(JoinedRoomUpdate {
1954+ timeline: Timeline {
1955+ events: vec![f.text_msg("old").event_id(old_id).into_event()],
1956+ ..Default::default()
1957+ },
1958+ ..Default::default()
1959+ })
1960+ .await
1961+ .unwrap();
1962+ room_event_cache
1963+ .handle_joined_room_update(JoinedRoomUpdate {
1964+ timeline: Timeline {
1965+ limited: true,
1966+ prev_batch: Some("cursor-gap".to_owned()),
1967+ events: vec![f.text_msg("cursor gap").event_id(cursor_gap_id).into_event()],
1968+ },
1969+ ..Default::default()
1970+ })
1971+ .await
1972+ .unwrap();
1973+ room_event_cache
1974+ .handle_joined_room_update(JoinedRoomUpdate {
1975+ timeline: Timeline {
1976+ limited: true,
1977+ events: vec![
1978+ f.text_msg("cursorless gap").event_id(cursorless_gap_id).into_event(),
1979+ ],
1980+ ..Default::default()
1981+ },
1982+ ..Default::default()
1983+ })
1984+ .await
1985+ .unwrap();
19401986
1941- // ev3 doesn't have an order with its previous position, since it's been
1942- // deduplicated.
1943- assert_eq!(
1944- room_linked_chunk.event_order(Position::new(ChunkIdentifier::new(1), 0)),
1945- None
1946- );
1947- }
1987+ let state = room_event_cache.inner.state.write().await.unwrap();
1988+ let candidate = state.history_candidate().await.unwrap();
1989+ let gaps: Vec<_> = candidate
1990+ .chunks()
1991+ .filter_map(|chunk| match chunk.content() {
1992+ ChunkContent::Gap(gap) => gap.history.clone(),
1993+ ChunkContent::Items(_) => None,
1994+ })
1995+ .collect();
1996+ assert_eq!(gaps.len(), 2);
1997+ assert_eq!(gaps[0].older_boundary.as_deref(), Some(old_id));
1998+ assert_eq!(gaps[0].newer_boundary.as_deref(), Some(cursor_gap_id));
1999+ assert_eq!(gaps[0].status, crate::event_cache::HistoryGapStatus::Idle);
2000+ assert_eq!(gaps[1].older_boundary.as_deref(), Some(cursor_gap_id));
2001+ assert_eq!(gaps[1].newer_boundary.as_deref(), Some(cursorless_gap_id));
2002+ assert_eq!(gaps[1].status, crate::event_cache::HistoryGapStatus::Unavailable);
19482003 }
19492004
19502005 #[async_test]
crates/matrix-sdk/src/event_cache/caches/room/pagination.rs+18−5
@@ -167,7 +167,12 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
167167
168168 // If any in-memory chunk is a gap, don't load more events, and let the caller
169169 // resolve the gap.
170- if let Some(prev_token) = state.room_linked_chunk().rgap().map(|gap| gap.token) {
170+ if let Some(prev_token) =
171+ state.room_linked_chunk().rchunks().find_map(|chunk| match chunk.content() {
172+ ChunkContent::Gap(gap) if gap.history.is_none() => Some(gap.token.clone()),
173+ _ => None,
174+ })
175+ {
171176 return Ok(LoadMoreEventsBackwardsOutcome::Gap {
172177 prev_token: Some(prev_token),
173178 waited_for_initial_prev_token: state.waited_for_initial_prev_token(),
@@ -255,7 +260,7 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
255260 let timeline_event_diffs = state.room_linked_chunk_mut().updates_as_vector_diffs();
256261
257262 Ok(match chunk_content {
258- ChunkContent::Gap(gap) => {
263+ ChunkContent::Gap(gap) if gap.history.is_none() => {
259264 trace!("reloaded chunk from disk (gap)");
260265
261266 LoadMoreEventsBackwardsOutcome::Gap {
@@ -264,6 +269,12 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
264269 }
265270 }
266271
272+ ChunkContent::Gap(_) => LoadMoreEventsBackwardsOutcome::Events {
273+ events: Vec::new(),
274+ timeline_event_diffs,
275+ reached_start: false,
276+ },
277+
267278 ChunkContent::Items(events) => {
268279 trace!(?reached_start, "reloaded chunk from disk ({} items)", events.len());
269280
@@ -314,7 +325,7 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
314325 reached_start: bool,
315326 generation: u64,
316327 ) -> Result<Option<BackPaginationOutcome>> {
317- let _state = self.state.read().await?;
328+ let state = self.state.read().await?;
318329 if self.generation.load(std::sync::atomic::Ordering::SeqCst) != generation {
319330 return Ok(None);
320331 }
@@ -322,6 +333,7 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
322333 self.update_sender.send(
323334 RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
324335 diffs: timeline_event_diffs,
336+ history_gaps: Some(super::history::gap_positions(state.room_linked_chunk())),
325337 origin: EventsOrigin::Cache,
326338 }),
327339 Some(RoomEventCacheGenericUpdate { room_id: self.room_id.clone() }),
@@ -353,7 +365,7 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
353365 let prev_gap_id = if let Some(token) = prev_token {
354366 // Find the corresponding gap in the in-memory linked chunk.
355367 let gap_chunk_id = state.room_linked_chunk().chunk_identifier(|chunk| {
356- matches!(chunk.content(), ChunkContent::Gap(Gap { token: prev_token }) if *prev_token == token)
368+ matches!(chunk.content(), ChunkContent::Gap(gap) if gap.history.is_none() && gap.token == token)
357369 });
358370
359371 if gap_chunk_id.is_none() {
@@ -417,7 +429,7 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
417429 // the inverted order; reorder them.
418430 let topo_ordered_events = events.iter().rev().cloned().collect::<Vec<_>>();
419431
420- let new_gap = new_token.map(|prev_token| Gap { token: prev_token });
432+ let new_gap = new_token.map(|prev_token| Gap { token: prev_token, history: None });
421433 let reached_start = state.room_linked_chunk_mut().push_backwards_pagination_events(
422434 prev_gap_id,
423435 new_gap,
@@ -448,6 +460,7 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
448460 self.update_sender.send(
449461 RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
450462 diffs: timeline_event_diffs,
463+ history_gaps: Some(super::history::gap_positions(state.room_linked_chunk())),
451464 origin: EventsOrigin::Pagination,
452465 }),
453466 Some(RoomEventCacheGenericUpdate { room_id: self.room_id.clone() }),
crates/matrix-sdk/src/event_cache/caches/room/state.rs+111−8
@@ -26,7 +26,7 @@ use matrix_sdk_base::{
2626 RoomInfoNotableUpdateReasons, apply_redaction, check_validity_of_replacement_events,
2727 deserialized_responses::{ThreadSummary, ThreadSummaryStatus},
2828 event_cache::{
29- Event, Gap,
29+ Event, Gap, HistoryGap, HistoryGapStatus,
3030 store::{EventCacheStoreLock, EventCacheStoreLockGuard, EventCacheStoreLockState},
3131 },
3232 linked_chunk::{
@@ -38,7 +38,7 @@ use matrix_sdk_base::{
3838 };
3939 use matrix_sdk_common::executor::spawn;
4040 use ruma::{
41- EventId, OwnedEventId, OwnedRoomId, OwnedUserId,
41+ EventId, OwnedEventId, OwnedRoomId, OwnedUserId, TransactionId,
4242 events::{
4343 AnySyncEphemeralRoomEvent, AnySyncMessageLikeEvent, AnySyncTimelineEvent,
4444 MessageLikeEventType,
@@ -422,6 +422,7 @@ impl<'a> lock::Reload for RoomEventCacheStateLockWriteGuard<'a> {
422422 self.state.update_sender.send(
423423 RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
424424 diffs,
425+ history_gaps: Some(super::history::gap_positions(&self.state.room_linked_chunk)),
425426 origin: EventsOrigin::Cache,
426427 }),
427428 Some(RoomEventCacheGenericUpdate { room_id: self.state.room_id.clone() }),
@@ -432,6 +433,10 @@ impl<'a> lock::Reload for RoomEventCacheStateLockWriteGuard<'a> {
432433 }
433434
434435 impl<'a> RoomEventCacheStateLockReadGuard<'a> {
436+ pub(crate) fn history_gap_positions(&self) -> Vec<super::HistoryGapPosition> {
437+ super::history::gap_positions(&self.state.room_linked_chunk)
438+ }
439+
435440 /// Return the subscriber count.
436441 pub fn subscriber_count(&self) -> &Arc<AtomicUsize> {
437442 &self.state.subscriber_count
@@ -784,7 +789,10 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
784789 events.retain(|event| event.event_id().is_none_or(|id| event_ids.insert(id)));
785790 let mut replacement = EventLinkedChunk::new();
786791 replacement.reset();
787- replacement.push_live_events(prev_batch.clone().map(|token| Gap { token }), &events);
792+ replacement.push_live_events(
793+ prev_batch.clone().map(|token| Gap { token, history: None }),
794+ &events,
795+ );
788796 let updates = replacement.store_updates().take();
789797 self.send_updates_to_store(updates).await?;
790798 let _ = replacement.updates_as_vector_diffs();
@@ -794,6 +802,61 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
794802 Ok(())
795803 }
796804
805+ pub(super) async fn history_candidate(&self) -> Result<EventLinkedChunk, EventCacheError> {
806+ let linked_chunk_id = LinkedChunkId::Room(&self.state.room_id);
807+ let (last, generator) = self.store.load_last_chunk(linked_chunk_id).await?;
808+ let metadata = self.store.load_all_chunks_metadata(linked_chunk_id).await?;
809+ let mut candidate = EventLinkedChunk::with_initial_linked_chunk(
810+ lazy_loader::from_last_chunk(last, generator)?,
811+ (!metadata.is_empty()).then_some(metadata),
812+ );
813+ loop {
814+ let first =
815+ candidate.chunks().next().expect("a linked chunk is never empty").identifier();
816+ let Some(previous) = self.store.load_previous_chunk(linked_chunk_id, first).await?
817+ else {
818+ break;
819+ };
820+ candidate.insert_new_chunk_as_first(previous)?;
821+ }
822+ let _ = candidate.store_updates().take();
823+ let _ = candidate.updates_as_vector_diffs();
824+ Ok(candidate)
825+ }
826+
827+ pub(crate) fn history_gap_positions(&self) -> Vec<super::HistoryGapPosition> {
828+ super::history::gap_positions(&self.state.room_linked_chunk)
829+ }
830+
831+ pub(super) async fn commit_history_candidate(
832+ &mut self,
833+ mut candidate: EventLinkedChunk,
834+ events: Vec<Event>,
835+ baseline_count: usize,
836+ ) -> Result<(), EventCacheError> {
837+ let use_incremental = self.state.room_linked_chunk.events().count() == baseline_count;
838+ self.send_updates_to_store(candidate.store_updates().take()).await?;
839+ self.state.room_linked_chunk = candidate;
840+ self.reset_pagination_status(false);
841+ let result =
842+ self.post_process_new_events(events, PostProcessingOrigin::Backpagination, None).await;
843+ let mut diffs = self.state.room_linked_chunk.updates_as_vector_diffs();
844+ if !use_incremental {
845+ let values =
846+ self.state.room_linked_chunk.events().map(|(_, event)| event.clone()).collect();
847+ diffs = vec![VectorDiff::Reset { values }];
848+ }
849+ self.state.update_sender.send(
850+ RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
851+ diffs,
852+ history_gaps: Some(super::history::gap_positions(&self.state.room_linked_chunk)),
853+ origin: EventsOrigin::Cache,
854+ }),
855+ Some(RoomEventCacheGenericUpdate { room_id: self.state.room_id.clone() }),
856+ );
857+ result
858+ }
859+
797860 fn reset_pagination_status(&self, hit_timeline_start: bool) {
798861 let mut status = self.state.pagination_status.write();
799862 self.state.generation.fetch_add(1, Ordering::SeqCst);
@@ -849,6 +912,7 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
849912 ephemeral_events: &[Raw<AnySyncEphemeralRoomEvent>],
850913 ) -> Result<(bool, Vec<VectorDiff<Event>>), EventCacheError> {
851914 let mut prev_batch = timeline.prev_batch.take();
915+ let had_prev_batch = prev_batch.is_some();
852916
853917 let DeduplicationOutcome {
854918 all_events: events,
@@ -882,9 +946,23 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
882946 prev_batch = None;
883947 }
884948
949+ let older_boundary = timeline
950+ .limited
951+ .then(|| self.state.room_linked_chunk.revents().find_map(|(_, event)| event.event_id()))
952+ .flatten();
953+ let newer_boundary =
954+ timeline.limited.then(|| events.iter().find_map(Event::event_id)).flatten();
955+ if older_boundary.is_some() && events.iter().any(|event| event.event_id() == older_boundary)
956+ {
957+ prev_batch = None;
958+ }
959+ let limited_discontinuity = older_boundary.is_some()
960+ && newer_boundary.is_some()
961+ && !events.iter().any(|event| event.event_id() == older_boundary);
885962 let has_new_gap = prev_batch.is_some();
963+ let has_new_discontinuity = has_new_gap || limited_discontinuity;
886964
887- if has_new_gap {
965+ if has_new_discontinuity {
888966 // Sad time: there's a gap, somewhere, in the timeline, and there's at least one
889967 // non-duplicated event. We don't know which threads might have gappy, so we
890968 // must invalidate them all :(
@@ -943,15 +1021,40 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
9431021 // events, because we are pushing all _new_ `events` at the back.
9441022 self.remove_events(in_memory_duplicated_event_ids, in_store_duplicated_event_ids).await?;
9451023
946- self.state
947- .room_linked_chunk
948- .push_live_events(prev_batch.map(|prev_token| Gap { token: prev_token }), &events);
1024+ let gap = prev_batch
1025+ .map(|token| Gap {
1026+ token,
1027+ history: limited_discontinuity.then(|| HistoryGap {
1028+ id: TransactionId::new(),
1029+ older_boundary: older_boundary.clone(),
1030+ newer_boundary: newer_boundary.clone(),
1031+ visited_cursors: Vec::new(),
1032+ status: if had_prev_batch {
1033+ HistoryGapStatus::Idle
1034+ } else {
1035+ HistoryGapStatus::Unavailable
1036+ },
1037+ }),
1038+ })
1039+ .or_else(|| {
1040+ limited_discontinuity.then(|| Gap {
1041+ token: String::new(),
1042+ history: Some(HistoryGap {
1043+ id: TransactionId::new(),
1044+ older_boundary,
1045+ newer_boundary,
1046+ visited_cursors: Vec::new(),
1047+ status: HistoryGapStatus::Unavailable,
1048+ }),
1049+ })
1050+ });
1051+ self.state.room_linked_chunk.push_live_events(gap, &events);
9491052
9501053 // Extract a new read receipt, if available.
9511054 let new_receipt = extract_read_receipt(ephemeral_events);
9521055 self.post_process_new_events(events, PostProcessingOrigin::Sync, new_receipt).await?;
9531056
954- if timeline.limited && has_new_gap {
1057+ if timeline.limited && has_new_discontinuity && !limited_discontinuity {
9551058 // If there was a previous batch token for a limited timeline, unload the chunks
9561059 // so it only contains the last one; otherwise, there might be a
9571060 // valid gap in between, and observers may not render it (yet).
crates/matrix-sdk/src/event_cache/caches/thread/mod.rs+3
@@ -130,6 +130,7 @@ impl ThreadEventCache {
130130 let _ = state.state.sender.send(TimelineVectorDiffs {
131131 diffs: updates_as_vector_diffs,
132132 origin: EventsOrigin::Cache,
133+ history_gaps: None,
133134 });
134135 }
135136
@@ -150,6 +151,7 @@ impl ThreadEventCache {
150151 let _ = state.state.sender.send(TimelineVectorDiffs {
151152 diffs: timeline_event_diffs,
152153 origin: EventsOrigin::Sync,
154+ history_gaps: None,
153155 });
154156 }
155157
@@ -181,6 +183,7 @@ impl ThreadEventCache {
181183 let _ = state.state.sender.send(TimelineVectorDiffs {
182184 diffs: timeline_event_diffs,
183185 origin: EventsOrigin::Sync,
186+ history_gaps: None,
184187 });
185188 }
186189
crates/matrix-sdk/src/event_cache/caches/thread/pagination.rs+7−6
@@ -218,7 +218,7 @@ impl PaginatedCache for ThreadEventCacheWrapper {
218218 // If the gap id is missing, it means that the gap disappeared during
219219 // pagination; in this case, early return to the caller.
220220 let Some(gap_id) = state.thread_linked_chunk().chunk_identifier(|chunk| {
221- matches!(chunk.content(), ChunkContent::Gap(Gap { token: prev_token }) if *prev_token == token)
221+ matches!(chunk.content(), ChunkContent::Gap(Gap { token: prev_token, .. }) if *prev_token == token)
222222 }) else {
223223 // The previous token has gone missing, so the timeline has been reset in the
224224 // meanwhile, but it's fine per this function's contract.
@@ -233,7 +233,7 @@ impl PaginatedCache for ThreadEventCacheWrapper {
233233 // This is a backwards pagination, so the events were returned in the reverse
234234 // topological order.
235235 let topo_ordered_events = events.iter().cloned().rev().collect::<Vec<_>>();
236- let new_gap = new_token.map(|token| Gap { token });
236+ let new_gap = new_token.map(|token| Gap { token, history: None });
237237
238238 let deduplication = state.filter_duplicate_events(topo_ordered_events);
239239
@@ -266,10 +266,11 @@ impl PaginatedCache for ThreadEventCacheWrapper {
266266
267267 if !updates.is_empty() {
268268 // Send the updates to the listeners.
269- let _ = state
270- .state
271- .sender
272- .send(TimelineVectorDiffs { diffs: updates, origin: EventsOrigin::Pagination });
269+ let _ = state.state.sender.send(TimelineVectorDiffs {
270+ diffs: updates,
271+ origin: EventsOrigin::Pagination,
272+ history_gaps: None,
273+ });
273274 }
274275
275276 Ok(Some(BackPaginationOutcome { reached_start, events }))
crates/matrix-sdk/src/event_cache/caches/thread/state.rs+5−2
@@ -124,8 +124,11 @@ impl<'a> lock::Reload for ThreadEventCacheStateLockWriteGuard<'a> {
124124 let diffs = self.state.thread_linked_chunk.updates_as_vector_diffs();
125125
126126 if !diffs.is_empty() {
127- let _ =
128- self.state.sender.send(TimelineVectorDiffs { diffs, origin: EventsOrigin::Cache });
127+ let _ = self.state.sender.send(TimelineVectorDiffs {
128+ diffs,
129+ origin: EventsOrigin::Cache,
130+ history_gaps: None,
131+ });
129132 }
130133
131134 Ok(())
crates/matrix-sdk/src/event_cache/mod.rs+3−2
@@ -70,8 +70,8 @@ pub use caches::{
7070 event_focused::EventFocusThreadMode,
7171 pagination::{BackPaginationOutcome, PaginationStatus},
7272 room::{
73- RoomEventCache, RoomEventCacheGenericUpdate, RoomEventCacheSubscriber,
74- RoomEventCacheUpdate, pagination::RoomPagination,
73+ HistoryGapFillOutcome, HistoryGapPosition, RecentHistoryOutcome, RoomEventCache, RoomEventCacheGenericUpdate,
74+ RoomEventCacheSubscriber, RoomEventCacheUpdate, pagination::RoomPagination,
7575 },
7676 thread::pagination::ThreadPagination,
7777 };
@@ -79,6 +79,7 @@ pub use caches::{
7979 pub use redecryptor::{DecryptionRetryRequest, RedecryptorReport};
8080
8181 pub use crate::event_cache::automatic_pagination::AutomaticPagination;
82+pub use matrix_sdk_base::event_cache::{HistoryGap, HistoryGapStatus};
8283
8384 /// An error observed in the [`EventCache`].
8485 #[derive(thiserror::Error, Clone, Debug)]
crates/matrix-sdk/src/event_cache/redecryptor.rs+2−1
@@ -57,7 +57,7 @@
5757 //! subscribes to [`RedecryptorReport`] stream.
5858 //!
5959 //! ```markdown
60-//!
60+//!
6161 //! .----------------------.
6262 //! | |
6363 //! | Beeb, boop! |
@@ -419,6 +419,7 @@ impl EventCache {
419419 RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
420420 diffs: state.room_linked_chunk_mut().updates_as_vector_diffs(),
421421 origin: EventsOrigin::Cache,
422+ history_gaps: Some(state.history_gap_positions()),
422423 }),
423424 Some(RoomEventCacheGenericUpdate { room_id: room_id.to_owned() }),
424425 );
crates/matrix-sdk/src/event_cache/tasks.rs+1
@@ -177,6 +177,7 @@ pub(super) async fn auto_shrink_linked_chunk_task(
177177 RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
178178 diffs,
179179 origin: EventsOrigin::Cache,
180+ history_gaps: Some(state.history_gap_positions()),
180181 }),
181182 None,
182183 );
crates/matrix-sdk/tests/integration/event_cache/mod.rs+1−1
@@ -1738,7 +1738,7 @@ async fn test_lazy_loading() {
17381738 previous: Some(ChunkIdentifier::new(0)),
17391739 new: ChunkIdentifier::new(1),
17401740 next: None,
1741- gap: Gap { token: "raclette".to_owned() },
1741+ gap: Gap { token: "raclette".to_owned(), history: None },
17421742 },
17431743 // chunk #3
17441744 Update::NewItemsChunk {
crates/matrix-sdk/tests/recent_history.rs+375
@@ -0,0 +1,375 @@
1+use std::time::Duration;
2+
3+use matrix_sdk::{
4+ event_cache::HistoryGapStatus,
5+ test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate},
6+};
7+use matrix_sdk_test::{JoinedRoomBuilder, async_test, event_factory::EventFactory};
8+use ruma::{event_id, room_id, user_id};
9+
10+fn event_factory(room_id: &ruma::RoomId) -> EventFactory {
11+ EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"))
12+}
13+
14+#[async_test]
15+async fn test_recent_history_is_bounded_and_keeps_the_retained_window() {
16+ let server = MatrixMockServer::new().await;
17+ let client = server.client_builder().build().await;
18+ client.event_cache().subscribe().unwrap();
19+ let room_id = room_id!("!history:example.org");
20+ let room = server.sync_joined_room(&client, room_id).await;
21+ let (cache, _handles) = room.event_cache().await.unwrap();
22+ let f = event_factory(room_id);
23+ cache
24+ .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
25+ .await
26+ .unwrap();
27+
28+ server
29+ .mock_room_messages()
30+ .match_limit(50)
31+ .ok(RoomMessagesResponseTemplate::default()
32+ .end_token("cursor-1")
33+ .events(vec![f.text_msg("recent-4").event_id(event_id!("$recent4"))]))
34+ .mock_once()
35+ .mount()
36+ .await;
37+ server
38+ .mock_room_messages()
39+ .match_from("cursor-1")
40+ .match_limit(50)
41+ .ok(RoomMessagesResponseTemplate::default()
42+ .end_token("cursor-2")
43+ .events(vec![f.text_msg("recent-3").event_id(event_id!("$recent3"))]))
44+ .mock_once()
45+ .mount()
46+ .await;
47+ server
48+ .mock_room_messages()
49+ .match_from("cursor-2")
50+ .match_limit(50)
51+ .ok(RoomMessagesResponseTemplate::default()
52+ .end_token("cursor-3")
53+ .events(vec![f.text_msg("recent-2").event_id(event_id!("$recent2"))]))
54+ .mock_once()
55+ .mount()
56+ .await;
57+ server
58+ .mock_room_messages()
59+ .match_from("cursor-3")
60+ .match_limit(50)
61+ .ok(RoomMessagesResponseTemplate::default()
62+ .end_token("cursor-4")
63+ .events(vec![f.text_msg("recent-1").event_id(event_id!("$recent1"))]))
64+ .mock_once()
65+ .mount()
66+ .await;
67+
68+ let outcome = cache.load_recent_history().await.unwrap();
69+ assert_eq!(outcome.pages, 4);
70+ assert_eq!(outcome.events, 4);
71+ assert_eq!(outcome.gap.as_ref().unwrap().status, HistoryGapStatus::Idle);
72+ let ids: Vec<_> =
73+ cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();
74+ assert_eq!(
75+ ids,
76+ vec![
77+ event_id!("$old"),
78+ event_id!("$recent1"),
79+ event_id!("$recent2"),
80+ event_id!("$recent3"),
81+ event_id!("$recent4"),
82+ ]
83+ );
84+ assert_eq!(cache.history_gaps().await.unwrap().len(), 1);
85+}
86+
87+#[async_test]
88+async fn test_recent_history_joins_at_the_retained_boundary() {
89+ let server = MatrixMockServer::new().await;
90+ let client = server.client_builder().build().await;
91+ client.event_cache().subscribe().unwrap();
92+ let room_id = room_id!("!history:example.org");
93+ let room = server.sync_joined_room(&client, room_id).await;
94+ let (cache, _handles) = room.event_cache().await.unwrap();
95+ let f = event_factory(room_id);
96+ cache
97+ .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
98+ .await
99+ .unwrap();
100+ server
101+ .mock_room_messages()
102+ .match_limit(50)
103+ .ok(RoomMessagesResponseTemplate::default().end_token("unused").events(vec![
104+ f.text_msg("recent").event_id(event_id!("$recent")),
105+ f.text_msg("old").event_id(event_id!("$old")),
106+ ]))
107+ .mock_once()
108+ .mount()
109+ .await;
110+
111+ let outcome = cache.load_recent_history().await.unwrap();
112+ assert_eq!(outcome.pages, 1);
113+ assert_eq!(outcome.gap.as_ref().unwrap().status, HistoryGapStatus::Joined);
114+ assert_eq!(
115+ cache
116+ .events()
117+ .await
118+ .unwrap()
119+ .iter()
120+ .filter_map(|event| event.event_id())
121+ .collect::<Vec<_>>(),
122+ vec![event_id!("$old"), event_id!("$recent")]
123+ );
124+}
125+
126+#[async_test]
127+async fn test_recent_history_keeps_sync_arrivals_during_a_fetch() {
128+ let server = MatrixMockServer::new().await;
129+ let client = server.client_builder().build().await;
130+ client.event_cache().subscribe().unwrap();
131+ let room_id = room_id!("!history:example.org");
132+ let room = server.sync_joined_room(&client, room_id).await;
133+ let (cache, _handles) = room.event_cache().await.unwrap();
134+ let f = event_factory(room_id);
135+ cache
136+ .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
137+ .await
138+ .unwrap();
139+ server
140+ .mock_room_messages()
141+ .match_limit(50)
142+ .ok(RoomMessagesResponseTemplate::default()
143+ .events(vec![
144+ f.text_msg("live").event_id(event_id!("$live")),
145+ f.text_msg("recent").event_id(event_id!("$recent")),
146+ ])
147+ .with_delay(Duration::from_millis(100)))
148+ .mock_once()
149+ .mount()
150+ .await;
151+
152+ let mut load = Box::pin(cache.load_recent_history());
153+ tokio::select! {
154+ biased;
155+ result = &mut load => panic!("request finished before sync {result:?}"),
156+ _ = tokio::time::sleep(Duration::from_millis(20)) => {}
157+ }
158+ server
159+ .sync_room(
160+ &client,
161+ JoinedRoomBuilder::new(room_id)
162+ .add_timeline_event(f.text_msg("live").event_id(event_id!("$live"))),
163+ )
164+ .await;
165+ load.await.unwrap();
166+
167+ let ids: Vec<_> =
168+ cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();
169+ assert_eq!(ids, vec![event_id!("$old"), event_id!("$recent"), event_id!("$live")]);
170+}
171+
172+#[async_test]
173+async fn test_recent_history_continues_empty_pages_and_pauses_on_a_cursor_cycle() {
174+ let server = MatrixMockServer::new().await;
175+ let client = server.client_builder().build().await;
176+ client.event_cache().subscribe().unwrap();
177+ let room_id = room_id!("!history:example.org");
178+ let room = server.sync_joined_room(&client, room_id).await;
179+ let (cache, _handles) = room.event_cache().await.unwrap();
180+ let f = event_factory(room_id);
181+ cache
182+ .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
183+ .await
184+ .unwrap();
185+ server
186+ .mock_room_messages()
187+ .match_limit(50)
188+ .ok(RoomMessagesResponseTemplate::default().end_token("cursor-a"))
189+ .mock_once()
190+ .mount()
191+ .await;
192+ server
193+ .mock_room_messages()
194+ .match_from("cursor-a")
195+ .match_limit(50)
196+ .ok(RoomMessagesResponseTemplate::default()
197+ .end_token("cursor-b")
198+ .events(vec![f.text_msg("recent").event_id(event_id!("$recent"))]))
199+ .mock_once()
200+ .mount()
201+ .await;
202+ server
203+ .mock_room_messages()
204+ .match_from("cursor-b")
205+ .match_limit(50)
206+ .ok(RoomMessagesResponseTemplate::default().end_token("cursor-a"))
207+ .mock_once()
208+ .mount()
209+ .await;
210+
211+ let outcome = cache.load_recent_history().await.unwrap();
212+ assert_eq!(outcome.pages, 3);
213+ assert_eq!(outcome.events, 1);
214+ assert_eq!(outcome.gap.as_ref().unwrap().status, HistoryGapStatus::Paused);
215+ assert_eq!(
216+ cache
217+ .events()
218+ .await
219+ .unwrap()
220+ .iter()
221+ .filter_map(|event| event.event_id())
222+ .collect::<Vec<_>>(),
223+ vec![event_id!("$old"), event_id!("$recent")]
224+ );
225+}
226+
227+#[async_test]
228+async fn test_concurrent_recent_history_calls_share_the_request_and_gap() {
229+ let server = MatrixMockServer::new().await;
230+ let client = server.client_builder().build().await;
231+ client.event_cache().subscribe().unwrap();
232+ let room_id = room_id!("!history:example.org");
233+ let room = server.sync_joined_room(&client, room_id).await;
234+ let (cache, _handles) = room.event_cache().await.unwrap();
235+ let f = event_factory(room_id);
236+ cache
237+ .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
238+ .await
239+ .unwrap();
240+ server
241+ .mock_room_messages()
242+ .match_limit(50)
243+ .ok(RoomMessagesResponseTemplate::default()
244+ .events(vec![f.text_msg("recent").event_id(event_id!("$recent"))])
245+ .with_delay(Duration::from_millis(100)))
246+ .expect(1)
247+ .mock_once()
248+ .mount()
249+ .await;
250+
251+ let (first, second) = tokio::join!(cache.load_recent_history(), cache.load_recent_history());
252+ let first = first.unwrap();
253+ let second = second.unwrap();
254+ assert_eq!(first.gap.as_ref().unwrap().id, second.gap.as_ref().unwrap().id);
255+ assert_eq!(first.pages, 1);
256+ assert_eq!(second.pages, 1);
257+ assert_eq!(first.events, 1);
258+ assert_eq!(second.events, 1);
259+}
260+
261+#[async_test]
262+async fn test_cancelling_one_recent_history_waiter_keeps_the_shared_request_alive() {
263+ let server = MatrixMockServer::new().await;
264+ let client = server.client_builder().build().await;
265+ client.event_cache().subscribe().unwrap();
266+ let room_id = room_id!("!history:example.org");
267+ let room = server.sync_joined_room(&client, room_id).await;
268+ let (cache, _handles) = room.event_cache().await.unwrap();
269+ let f = event_factory(room_id);
270+ cache
271+ .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
272+ .await
273+ .unwrap();
274+ server
275+ .mock_room_messages()
276+ .match_limit(50)
277+ .ok(RoomMessagesResponseTemplate::default()
278+ .events(vec![f.text_msg("recent").event_id(event_id!("$recent"))])
279+ .with_delay(Duration::from_millis(100)))
280+ .expect(1)
281+ .mock_once()
282+ .mount()
283+ .await;
284+
285+ let mut first = Box::pin(cache.load_recent_history());
286+ tokio::select! {
287+ biased;
288+ result = &mut first => panic!("request finished before second waiter {result:?}"),
289+ _ = tokio::time::sleep(Duration::from_millis(20)) => {}
290+ }
291+ let second_cache = cache.clone();
292+ let second = tokio::spawn(async move { second_cache.load_recent_history().await });
293+ tokio::time::sleep(Duration::from_millis(20)).await;
294+ drop(first);
295+ let outcome = second.await.unwrap().unwrap();
296+ assert_eq!(outcome.pages, 1);
297+ assert_eq!(outcome.gap.unwrap().status, HistoryGapStatus::Unavailable);
298+ assert_eq!(
299+ cache
300+ .events()
301+ .await
302+ .unwrap()
303+ .iter()
304+ .filter_map(|event| event.event_id())
305+ .collect::<Vec<_>>(),
306+ vec![event_id!("$old"), event_id!("$recent")]
307+ );
308+}
309+
310+#[async_test]
311+async fn test_cancelling_all_recent_history_waiters_allows_a_fresh_retry() {
312+ let server = MatrixMockServer::new().await;
313+ let client = server.client_builder().build().await;
314+ client.event_cache().subscribe().unwrap();
315+ let room_id = room_id!("!history:example.org");
316+ let room = server.sync_joined_room(&client, room_id).await;
317+ let (cache, _handles) = room.event_cache().await.unwrap();
318+ let f = event_factory(room_id);
319+ cache
320+ .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
321+ .await
322+ .unwrap();
323+ server
324+ .mock_room_messages()
325+ .match_limit(50)
326+ .ok(RoomMessagesResponseTemplate::default()
327+ .events(vec![f.text_msg("discarded").event_id(event_id!("$discarded"))])
328+ .with_delay(Duration::from_millis(300)))
329+ .expect(1)
330+ .mock_once()
331+ .mount()
332+ .await;
333+ server
334+ .mock_room_messages()
335+ .match_limit(50)
336+ .ok(RoomMessagesResponseTemplate::default()
337+ .events(vec![f.text_msg("recent").event_id(event_id!("$recent"))]))
338+ .expect(1)
339+ .mock_once()
340+ .mount()
341+ .await;
342+
343+ let mut abandoned = Box::pin(cache.load_recent_history());
344+ tokio::select! {
345+ biased;
346+ result = &mut abandoned => panic!("request finished before cancellation {result:?}"),
347+ _ = tokio::time::sleep(Duration::from_millis(20)) => {}
348+ }
349+ drop(abandoned);
350+ tokio::time::sleep(Duration::from_millis(30)).await;
351+ assert_eq!(
352+ cache
353+ .events()
354+ .await
355+ .unwrap()
356+ .iter()
357+ .filter_map(|event| event.event_id())
358+ .collect::<Vec<_>>(),
359+ vec![event_id!("$old")]
360+ );
361+
362+ let outcome = cache.load_recent_history().await.unwrap();
363+ assert_eq!(outcome.pages, 1);
364+ assert_eq!(outcome.gap.unwrap().status, HistoryGapStatus::Unavailable);
365+ assert_eq!(
366+ cache
367+ .events()
368+ .await
369+ .unwrap()
370+ .iter()
371+ .filter_map(|event| event.event_id())
372+ .collect::<Vec<_>>(),
373+ vec![event_id!("$old"), event_id!("$recent")]
374+ );
375+}
crates/matrix-sdk/tests/segmented_history.rs+472
@@ -0,0 +1,472 @@
1+use std::{
2+ sync::{Arc, atomic::Ordering},
3+ time::Duration,
4+};
5+
6+use matrix_sdk::{
7+ event_cache::{HistoryGapFillOutcome, HistoryGapStatus},
8+ store::StoreConfig,
9+ test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate},
10+};
11+use matrix_sdk_base::event_cache::store::MemoryStore;
12+use matrix_sdk_common::cross_process_lock::CrossProcessLockConfig;
13+use matrix_sdk_test::{JoinedRoomBuilder, async_test, event_factory::EventFactory};
14+use ruma::{event_id, room_id, user_id};
15+
16+#[path = "history_reseed/failing_store.rs"]
17+mod failing_store;
18+
19+#[async_test]
20+async fn test_segmented_history_resumes_after_restart_and_joins_only_at_boundary() {
21+ let server = MatrixMockServer::new().await;
22+ let config = StoreConfig::new(CrossProcessLockConfig::multi_process("segmented-history"))
23+ .state_store(matrix_sdk_base::store::MemoryStore::new())
24+ .event_cache_store(Arc::new(MemoryStore::new()));
25+ let room_id = room_id!("!history:example.org");
26+ let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
27+ let gap_id;
28+ {
29+ let client = server
30+ .client_builder()
31+ .on_builder(|builder| builder.store_config(config.clone()))
32+ .build()
33+ .await;
34+ client.event_cache().subscribe().unwrap();
35+ let room = server.sync_joined_room(&client, room_id).await;
36+ let (cache, _handles) = room.event_cache().await.unwrap();
37+ cache
38+ .reseed(
39+ vec![f.text_msg("retained").event_id(event_id!("$old")).into_event()],
40+ Some("old-edge".into()),
41+ )
42+ .await
43+ .unwrap();
44+ let gap = cache
45+ .append_history_section(
46+ vec![f.text_msg("recent").event_id(event_id!("$new")).into_event()],
47+ Some("recent-edge".into()),
48+ )
49+ .await
50+ .unwrap()
51+ .unwrap();
52+ gap_id = gap.id.clone();
53+ let mut expected_gap = gap.clone();
54+ expected_gap.visited_cursors.push("recent-edge".into());
55+ server
56+ .mock_room_messages()
57+ .match_from("recent-edge")
58+ .match_limit(50)
59+ .ok(RoomMessagesResponseTemplate::default().end_token("after-empty"))
60+ .mock_once()
61+ .mount()
62+ .await;
63+ assert_eq!(
64+ cache.fill_history_gap_once(gap.id.clone()).await.unwrap(),
65+ HistoryGapFillOutcome::Updated { gap: expected_gap.clone(), events: 0 }
66+ );
67+ assert_eq!(cache.history_gaps().await.unwrap(), vec![expected_gap]);
68+ assert_eq!(cache.events().await.unwrap().len(), 2);
69+ }
70+ let client =
71+ server.client_builder().on_builder(|builder| builder.store_config(config)).build().await;
72+ client.event_cache().subscribe().unwrap();
73+ let room = client.get_room(room_id).unwrap();
74+ let (cache, _handles) = room.event_cache().await.unwrap();
75+ assert_eq!(cache.history_gaps().await.unwrap()[0].id, gap_id);
76+ cache.load_retained_history().await.unwrap();
77+ let (restored, gaps) = cache.history_snapshot().await.unwrap();
78+ assert_eq!(
79+ restored.iter().filter_map(|event| event.event_id()).collect::<Vec<_>>(),
80+ vec![event_id!("$old"), event_id!("$new")]
81+ );
82+ assert_eq!(gaps[0].gap.id, gap_id);
83+ server
84+ .mock_room_messages()
85+ .match_from("after-empty")
86+ .match_limit(50)
87+ .ok(RoomMessagesResponseTemplate::default()
88+ .end_token("bridge-edge")
89+ .events(vec![f.text_msg("middle").event_id(event_id!("$middle"))]))
90+ .mock_once()
91+ .mount()
92+ .await;
93+ let HistoryGapFillOutcome::Updated { gap, events } =
94+ cache.fill_history_gap_once(gap_id.clone()).await.unwrap()
95+ else {
96+ panic!()
97+ };
98+ assert_eq!(events, 1);
99+ assert_eq!(gap.status, HistoryGapStatus::Idle);
100+ server
101+ .mock_room_messages()
102+ .match_from("bridge-edge")
103+ .ok(RoomMessagesResponseTemplate::default()
104+ .end_token("irrelevant")
105+ .events(vec![f.text_msg("retained").event_id(event_id!("$old"))]))
106+ .mock_once()
107+ .mount()
108+ .await;
109+ let HistoryGapFillOutcome::Updated { gap, events } =
110+ cache.fill_history_gap_once(gap_id.clone()).await.unwrap()
111+ else {
112+ panic!()
113+ };
114+ assert_eq!(gap.status, HistoryGapStatus::Joined);
115+ assert_eq!(events, 0);
116+ assert_eq!(
117+ cache.fill_history_gap_once(gap_id).await.unwrap(),
118+ HistoryGapFillOutcome::Updated { gap, events: 0 }
119+ );
120+ let ids: Vec<_> =
121+ cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();
122+ assert_eq!(ids, vec![event_id!("$old"), event_id!("$middle"), event_id!("$new")]);
123+}
124+
125+#[async_test]
126+async fn test_subscribe_with_history_reports_atomic_gap_snapshot_and_updates() {
127+ let server = MatrixMockServer::new().await;
128+ let client = server.client_builder().build().await;
129+ client.event_cache().subscribe().unwrap();
130+ let room_id = room_id!("!history:example.org");
131+ let room = server.sync_joined_room(&client, room_id).await;
132+ let (cache, _handles) = room.event_cache().await.unwrap();
133+ let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
134+ cache
135+ .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
136+ .await
137+ .unwrap();
138+ cache
139+ .append_history_section(
140+ vec![f.reaction(event_id!("$old"), "👍").event_id(event_id!("$hidden")).into_event()],
141+ Some("hidden-cursor".into()),
142+ )
143+ .await
144+ .unwrap()
145+ .unwrap();
146+ cache
147+ .append_history_section(
148+ vec![f.text_msg("new").event_id(event_id!("$new")).into_event()],
149+ Some("new-cursor".into()),
150+ )
151+ .await
152+ .unwrap()
153+ .unwrap();
154+
155+ let (snapshot, gaps, mut subscriber) = cache.subscribe_with_history().await.unwrap();
156+ let snapshot_ids: Vec<_> = snapshot.iter().filter_map(|event| event.event_id()).collect();
157+ assert_eq!(snapshot_ids, vec![event_id!("$old"), event_id!("$hidden"), event_id!("$new")]);
158+ assert_eq!(gaps[0].next_event_id.as_deref(), Some(event_id!("$hidden")));
159+ let gap_id = gaps[0].gap.id.clone();
160+
161+ server
162+ .mock_room_messages()
163+ .match_from("hidden-cursor")
164+ .ok(RoomMessagesResponseTemplate::default()
165+ .end_token("next-cursor")
166+ .events(vec![f.text_msg("middle").event_id(event_id!("$middle"))]))
167+ .mock_once()
168+ .mount()
169+ .await;
170+ cache.fill_history_gap_once(gap_id.clone()).await.unwrap();
171+
172+ let update =
173+ tokio::time::timeout(Duration::from_secs(2), subscriber.recv()).await.unwrap().unwrap();
174+ let matrix_sdk::event_cache::RoomEventCacheUpdate::UpdateTimelineEvents(update) = update else {
175+ panic!("expected timeline update")
176+ };
177+ let gaps = update.history_gaps.unwrap();
178+ assert_eq!(gaps[0].gap.id, gap_id);
179+ assert_eq!(gaps[0].next_event_id.as_deref(), Some(event_id!("$middle")));
180+ let mut visible: imbl::Vector<_> = snapshot.into();
181+ for diff in update.diffs {
182+ diff.apply(&mut visible);
183+ }
184+ let ids: Vec<_> = visible.iter().filter_map(|event| event.event_id()).collect();
185+ assert_eq!(
186+ ids,
187+ vec![event_id!("$old"), event_id!("$middle"), event_id!("$hidden"), event_id!("$new")]
188+ );
189+}
190+
191+#[async_test]
192+async fn test_segmented_history_keeps_unavailable_intervals_and_independent_cursors() {
193+ let server = MatrixMockServer::new().await;
194+ let client = server.client_builder().build().await;
195+ client.event_cache().subscribe().unwrap();
196+ let room_id = room_id!("!history:example.org");
197+ let room = server.sync_joined_room(&client, room_id).await;
198+ let (cache, _handles) = room.event_cache().await.unwrap();
199+ let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
200+ cache
201+ .reseed(
202+ vec![f.text_msg("old").event_id(event_id!("$old")).into_event()],
203+ Some("old-edge".into()),
204+ )
205+ .await
206+ .unwrap();
207+ let first = cache
208+ .append_history_section(
209+ vec![f.text_msg("middle").event_id(event_id!("$middle")).into_event()],
210+ Some("first".into()),
211+ )
212+ .await
213+ .unwrap()
214+ .unwrap();
215+ let second = cache
216+ .append_history_section(
217+ vec![f.text_msg("new").event_id(event_id!("$new")).into_event()],
218+ Some("second".into()),
219+ )
220+ .await
221+ .unwrap()
222+ .unwrap();
223+ server
224+ .mock_room_messages()
225+ .match_from("first")
226+ .ok(RoomMessagesResponseTemplate::default().end_token("first"))
227+ .mock_once()
228+ .mount()
229+ .await;
230+ let HistoryGapFillOutcome::Updated { gap, .. } =
231+ cache.fill_history_gap_once(first.id.clone()).await.unwrap()
232+ else {
233+ panic!()
234+ };
235+ assert_eq!(gap.status, HistoryGapStatus::Paused);
236+ server
237+ .mock_room_messages()
238+ .match_from("second")
239+ .ok(RoomMessagesResponseTemplate::default())
240+ .mock_once()
241+ .mount()
242+ .await;
243+ let HistoryGapFillOutcome::Updated { gap, .. } =
244+ cache.fill_history_gap_once(second.id).await.unwrap()
245+ else {
246+ panic!()
247+ };
248+ assert_eq!(gap.status, HistoryGapStatus::Unavailable);
249+ assert_eq!(cache.history_gaps().await.unwrap()[0].status, HistoryGapStatus::Paused);
250+ assert_eq!(cache.events().await.unwrap().len(), 3);
251+}
252+
253+#[async_test]
254+async fn test_segmented_history_failed_commit_keeps_events_and_cursor_retryable() {
255+ let server = MatrixMockServer::new().await;
256+ let store = Arc::new(failing_store::FailingStore::default());
257+ let config = StoreConfig::new(CrossProcessLockConfig::multi_process("history-write-failure"))
258+ .event_cache_store(store.clone());
259+ let client =
260+ server.client_builder().on_builder(|builder| builder.store_config(config)).build().await;
261+ client.event_cache().subscribe().unwrap();
262+ let room_id = room_id!("!history:example.org");
263+ let room = server.sync_joined_room(&client, room_id).await;
264+ let (cache, _handles) = room.event_cache().await.unwrap();
265+ let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
266+ cache
267+ .reseed(
268+ vec![f.text_msg("old").event_id(event_id!("$old")).into_event()],
269+ Some("old-edge".into()),
270+ )
271+ .await
272+ .unwrap();
273+ store.fail_writes.store(true, Ordering::SeqCst);
274+ assert!(
275+ cache
276+ .append_history_section(
277+ vec![f.text_msg("new").event_id(event_id!("$new")).into_event()],
278+ Some("first".into()),
279+ )
280+ .await
281+ .is_err()
282+ );
283+ assert_eq!(cache.events().await.unwrap().len(), 1);
284+ assert!(cache.history_gaps().await.unwrap().is_empty());
285+ store.fail_writes.store(false, Ordering::SeqCst);
286+ let gap = cache
287+ .append_history_section(
288+ vec![f.text_msg("new").event_id(event_id!("$new")).into_event()],
289+ Some("first".into()),
290+ )
291+ .await
292+ .unwrap()
293+ .unwrap();
294+ server
295+ .mock_room_messages()
296+ .match_from("first")
297+ .ok(RoomMessagesResponseTemplate::default()
298+ .end_token("next")
299+ .events(vec![f.text_msg("inserted").event_id(event_id!("$inserted"))]))
300+ .expect(2)
301+ .mount()
302+ .await;
303+ store.fail_writes.store(true, Ordering::SeqCst);
304+ assert!(cache.fill_history_gap_once(gap.id.clone()).await.is_err());
305+ assert_eq!(cache.events().await.unwrap().len(), 2);
306+ assert_eq!(cache.history_gaps().await.unwrap(), vec![gap.clone()]);
307+ store.fail_writes.store(false, Ordering::SeqCst);
308+ let HistoryGapFillOutcome::Updated { events, .. } =
309+ cache.fill_history_gap_once(gap.id).await.unwrap()
310+ else {
311+ panic!()
312+ };
313+ assert_eq!(events, 1);
314+ assert_eq!(cache.events().await.unwrap().len(), 3);
315+}
316+
317+#[async_test]
318+async fn test_segmented_history_request_failure_preserves_existing_progress() {
319+ let server = MatrixMockServer::new().await;
320+ let client = server.client_builder().build().await;
321+ client.event_cache().subscribe().unwrap();
322+ let room_id = room_id!("!history:example.org");
323+ let room = server.sync_joined_room(&client, room_id).await;
324+ let (cache, _handles) = room.event_cache().await.unwrap();
325+ let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
326+ cache
327+ .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
328+ .await
329+ .unwrap();
330+ let gap = cache
331+ .append_history_section(
332+ vec![f.text_msg("new").event_id(event_id!("$new")).into_event()],
333+ Some("first".into()),
334+ )
335+ .await
336+ .unwrap()
337+ .unwrap();
338+ server.mock_room_messages().match_from("first").error_unrecognized().mock_once().mount().await;
339+ assert!(cache.fill_history_gap_once(gap.id.clone()).await.is_err());
340+ assert!(matches!(
341+ cache.history_gaps().await.unwrap()[0].status,
342+ HistoryGapStatus::Failed { .. }
343+ ));
344+ assert_eq!(cache.events().await.unwrap().len(), 2);
345+ server
346+ .mock_room_messages()
347+ .match_from("first")
348+ .ok(RoomMessagesResponseTemplate::default()
349+ .events(vec![f.text_msg("old").event_id(event_id!("$old"))]))
350+ .mock_once()
351+ .mount()
352+ .await;
353+ let HistoryGapFillOutcome::Updated { gap, .. } =
354+ cache.fill_history_gap_once(gap.id).await.unwrap()
355+ else {
356+ panic!()
357+ };
358+ assert_eq!(gap.status, HistoryGapStatus::Joined);
359+}
360+
361+#[async_test]
362+async fn test_segmented_history_cancellation_keeps_the_saved_cursor() {
363+ let server = MatrixMockServer::new().await;
364+ let client = server.client_builder().build().await;
365+ client.event_cache().subscribe().unwrap();
366+ let room_id = room_id!("!history:example.org");
367+ let room = server.sync_joined_room(&client, room_id).await;
368+ let (cache, _handles) = room.event_cache().await.unwrap();
369+ let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
370+ cache
371+ .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
372+ .await
373+ .unwrap();
374+ let gap = cache
375+ .append_history_section(
376+ vec![f.text_msg("new").event_id(event_id!("$new")).into_event()],
377+ Some("saved".into()),
378+ )
379+ .await
380+ .unwrap()
381+ .unwrap();
382+ server
383+ .mock_room_messages()
384+ .match_from("saved")
385+ .ok(RoomMessagesResponseTemplate::default()
386+ .end_token("uncommitted")
387+ .with_delay(Duration::from_millis(300)))
388+ .mock_once()
389+ .mount()
390+ .await;
391+ let mut fill = Box::pin(cache.fill_history_gap_once(gap.id.clone()));
392+ tokio::select! {
393+ biased;
394+ result = &mut fill => panic!("request finished before cancellation {result:?}"),
395+ _ = tokio::time::sleep(Duration::from_millis(20)) => {}
396+ }
397+ assert_eq!(
398+ cache.fill_history_gap_once(gap.id.clone()).await.unwrap(),
399+ HistoryGapFillOutcome::Busy
400+ );
401+ drop(fill);
402+ assert_eq!(cache.history_gaps().await.unwrap(), vec![gap.clone()]);
403+ server
404+ .mock_room_messages()
405+ .match_from("saved")
406+ .ok(RoomMessagesResponseTemplate::default()
407+ .events(vec![f.text_msg("old").event_id(event_id!("$old"))]))
408+ .mock_once()
409+ .mount()
410+ .await;
411+ let HistoryGapFillOutcome::Updated { gap, .. } =
412+ cache.fill_history_gap_once(gap.id).await.unwrap()
413+ else {
414+ panic!()
415+ };
416+ assert_eq!(gap.status, HistoryGapStatus::Joined);
417+}
418+
419+#[async_test]
420+async fn test_segmented_history_preserves_sync_arrivals_during_a_fill() {
421+ let server = MatrixMockServer::new().await;
422+ let client = server.client_builder().build().await;
423+ client.event_cache().subscribe().unwrap();
424+ let room_id = room_id!("!history:example.org");
425+ let room = server.sync_joined_room(&client, room_id).await;
426+ let (cache, _handles) = room.event_cache().await.unwrap();
427+ let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
428+ cache
429+ .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
430+ .await
431+ .unwrap();
432+ let gap = cache
433+ .append_history_section(
434+ vec![f.text_msg("new").event_id(event_id!("$new")).into_event()],
435+ Some("saved".into()),
436+ )
437+ .await
438+ .unwrap()
439+ .unwrap();
440+ let (_, mut updates) = cache.subscribe().await.unwrap();
441+ server
442+ .mock_room_messages()
443+ .match_from("saved")
444+ .ok(RoomMessagesResponseTemplate::default()
445+ .end_token("next")
446+ .events(vec![f.text_msg("middle").event_id(event_id!("$middle"))])
447+ .with_delay(Duration::from_millis(100)))
448+ .mock_once()
449+ .mount()
450+ .await;
451+ let mut fill = Box::pin(cache.fill_history_gap_once(gap.id));
452+ tokio::select! {
453+ biased;
454+ result = &mut fill => panic!("request finished before sync {result:?}"),
455+ _ = tokio::time::sleep(Duration::from_millis(20)) => {}
456+ }
457+ server
458+ .sync_room(
459+ &client,
460+ JoinedRoomBuilder::new(room_id)
461+ .add_timeline_event(f.text_msg("live").event_id(event_id!("$live"))),
462+ )
463+ .await;
464+ tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
465+ fill.await.unwrap();
466+ let ids: Vec<_> =
467+ cache.events().await.unwrap().iter().filter_map(|event| event.event_id()).collect();
468+ assert_eq!(
469+ ids,
470+ vec![event_id!("$old"), event_id!("$middle"), event_id!("$new"), event_id!("$live")]
471+ );
472+}