matrix-rust-sdk

public

Star0

Fork 0

Watch0

9 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `ef3c294b31cc`

tarball

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

9 files changed+578-29

event_cache: add durable room history reseeding

Modified file: crates/matrix-sdk/Cargo.toml+4 additions@@ -209,5 +209,9 @@ required-features = ["testing"]
 name = "response_commit"
 required-features = ["testing", "e2e-encryption"]
 
+[[test]]
+name = "history_reseed"
+required-features = ["testing"]
+
 [lints]
 workspace = true

Modified file: crates/matrix-sdk/src/event_cache/caches/pagination.rs+45 additions−22 deletions@@ -144,24 +144,31 @@ where
             }
         }
 
+        let generation = self.cache.generation();
         let reset_status_on_drop_guard = ResetStatusOnDrop {
             prev_status: Some(status_guard.clone()),
-            pagination_status: status_observable.clone(),
+            cache: self.cache.clone(),
+            generation,
         };
 
         let this = self.clone();
 
         let fut: Pin<Box<dyn SharedPaginationFuture>> = Box::pin(async move {
-            match this.paginate_backwards_impl(batch_size).await? {
+            match this.paginate_backwards_impl(batch_size, generation).await? {
                 Some(outcome) => {
+                    let mut status = this.cache.status().write();
+                    if this.cache.generation() != generation {
+                        return Ok(None);
+                    }
                     // Back-pagination's over and successful, don't reset the status to the previous
                     // value.
                     reset_status_on_drop_guard.disarm();
 
                     // Notify subscribers that pagination ended.
-                    this.cache.status().set(SharedPaginationStatus::Idle {
-                        hit_timeline_start: outcome.reached_start,
-                    });
+                    ObservableWriteGuard::set(
+                        &mut status,
+                        SharedPaginationStatus::Idle { hit_timeline_start: outcome.reached_start },
+                    );
 
                     Ok(Some(outcome))
                 }
@@ -207,6 +214,7 @@ where
     async fn paginate_backwards_impl(
         &self,
         batch_size: u16,
+        generation: u64,
     ) -> Result<Option<BackPaginationOutcome>> {
         // A linked chunk might not be entirely loaded (if it's been lazy-loaded). Try
         // to load from disk/storage first, then from network if disk/storage indicated
@@ -250,7 +258,9 @@ where
                     }
 
                     // We have a gap, so resolve it with a network back-pagination.
-                    return self.paginate_backwards_with_network(batch_size, prev_token).await;
+                    return self
+                        .paginate_backwards_with_network(batch_size, prev_token, generation)
+                        .await;
                 }
 
                 LoadMoreEventsBackwardsOutcome::StartOfTimeline => {
@@ -262,15 +272,15 @@ where
                     timeline_event_diffs,
                     reached_start,
                 } => {
-                    return Ok(Some(
-                        self.cache
-                            .conclude_backwards_pagination_from_disk(
-                                events,
-                                timeline_event_diffs,
-                                reached_start,
-                            )
-                            .await,
-                    ));
+                    return self
+                        .cache
+                        .conclude_backwards_pagination_from_disk(
+                            events,
+                            timeline_event_diffs,
+                            reached_start,
+                            generation,
+                        )
+                        .await;
                 }
             }
         }
@@ -285,6 +295,7 @@ where
         &self,
         batch_size: u16,
         prev_token: Option<String>,
+        generation: u64,
     ) -> Result<Option<BackPaginationOutcome>> {
         let Some((events, new_token)) =
             self.cache.paginate_backwards_with_network(batch_size, &prev_token).await?
@@ -296,7 +307,9 @@ where
             }));
         };
 
-        self.cache.conclude_backwards_pagination_from_network(events, prev_token, new_token).await
+        self.cache
+            .conclude_backwards_pagination_from_network(events, prev_token, new_token, generation)
+            .await
     }
 }
 
@@ -340,6 +353,10 @@ pub(in super::super) enum SharedPaginationStatus {
 }
 
 pub(in super::super) trait PaginatedCache {
+    fn generation(&self) -> u64 {
+        0
+    }
+
     fn status(&self) -> &SharedObservable<SharedPaginationStatus>;
 
     fn load_more_events_backwards(
@@ -363,13 +380,15 @@ pub(in super::super) trait PaginatedCache {
         events: Vec<Event>,
         timeline_event_diffs: Vec<VectorDiff<Event>>,
         reached_start: bool,
-    ) -> impl Future<Output = BackPaginationOutcome> + SendOutsideWasm;
+        generation: u64,
+    ) -> impl Future<Output = Result<Option<BackPaginationOutcome>>> + SendOutsideWasm;
 
     fn conclude_backwards_pagination_from_network(
         &self,
         events: Vec<Event>,
         prev_token: Option<String>,
         new_token: Option<String>,
+        generation: u64,
     ) -> impl Future<Output = Result<Option<BackPaginationOutcome>>> + SendOutsideWasm;
 }
 
@@ -390,22 +409,26 @@ pub enum PaginationStatus {
 
 /// Small RAII guard to reset the pagination status on drop, if not disarmed in
 /// the meanwhile.
-struct ResetStatusOnDrop {
+struct ResetStatusOnDrop<C: PaginatedCache> {
     prev_status: Option<SharedPaginationStatus>,
-    pagination_status: SharedObservable<SharedPaginationStatus>,
+    cache: C,
+    generation: u64,
 }
 
-impl ResetStatusOnDrop {
+impl<C: PaginatedCache> ResetStatusOnDrop<C> {
     /// Make the RAII guard have no effect.
     fn disarm(mut self) {
         self.prev_status = None;
     }
 }
 
-impl Drop for ResetStatusOnDrop {
+impl<C: PaginatedCache> Drop for ResetStatusOnDrop<C> {
     fn drop(&mut self) {
         if let Some(status) = self.prev_status.take() {
-            let _ = self.pagination_status.set(status);
+            let mut guard = self.cache.status().write();
+            if self.cache.generation() == self.generation {
+                ObservableWriteGuard::set(&mut guard, status);
+            }
         }
     }
 }

Modified file: crates/matrix-sdk/src/event_cache/caches/room/mod.rs+58 additions−3 deletions@@ -20,10 +20,13 @@ mod updates;
 use std::{
     collections::BTreeMap,
     fmt,
-    sync::{Arc, atomic::Ordering},
+    sync::{
+        Arc,
+        atomic::{AtomicU64, Ordering},
+    },
 };
 
-use eyeball::SharedObservable;
+use eyeball::{ObservableWriteGuard, SharedObservable};
 use matrix_sdk_base::{
     deserialized_responses::AmbiguityChange,
     event_cache::Event,
@@ -336,7 +339,16 @@ impl RoomEventCache {
     /// storage.
     pub async fn clear(&self) -> Result<()> {
         // Clear the linked chunk and persisted storage.
-        let updates_as_vector_diffs = self.inner.state.write().await?.reset().await?;
+        let mut state = self.inner.state.write().await?;
+        let updates_as_vector_diffs = state.reset().await?;
+        {
+            let mut status = self.inner.shared_pagination_status.write();
+            self.inner.generation.fetch_add(1, Ordering::SeqCst);
+            ObservableWriteGuard::set(
+                &mut status,
+                SharedPaginationStatus::Idle { hit_timeline_start: false },
+            );
+        }
 
         // Notify observers about the update.
         self.inner.update_sender.send(
@@ -350,6 +362,47 @@ impl RoomEventCache {
         Ok(())
     }
 
+    #[doc = include_str!("reseed.md")]
+    pub async fn reseed(&self, events: Vec<Event>, prev_batch: Option<String>) -> Result<()> {
+        let cache = self.clone();
+        matrix_sdk_common::executor::spawn(async move {
+            let mut state = cache.inner.state.write().await?;
+            state.reseed(events, prev_batch.clone()).await?;
+            {
+                let mut status = cache.inner.shared_pagination_status.write();
+                cache.inner.generation.fetch_add(1, Ordering::SeqCst);
+                ObservableWriteGuard::set(
+                    &mut status,
+                    SharedPaginationStatus::Idle { hit_timeline_start: prev_batch.is_none() },
+                );
+            }
+            let events =
+                state.room_linked_chunk().events().map(|(_, event)| event.clone()).collect();
+            let result = async {
+                state.clear_loaded_threads().await?;
+                state
+                    .post_process_new_events(events, PostProcessingOrigin::Backpagination, None)
+                    .await
+            }
+            .await;
+            let items =
+                state.room_linked_chunk().events().map(|(_, event)| event.clone()).collect();
+            cache.inner.update_sender.send(
+                RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
+                    diffs: vec![eyeball_im::VectorDiff::Reset { values: items }],
+                    origin: EventsOrigin::Cache,
+                }),
+                Some(RoomEventCacheGenericUpdate { room_id: cache.inner.room_id.clone() }),
+            );
+            if prev_batch.is_some() {
+                cache.inner.pagination_batch_token_notifier.notify_waiters();
+            }
+            result
+        })
+        .await
+        .expect("joining history reseed failed")
+    }
+
     /// Return a reference to the state.
     pub(in super::super) fn state(&self) -> &LockedRoomEventCacheState {
         &self.inner.state
@@ -423,6 +476,7 @@ impl RoomEventCache {
 
 /// The (non-cloneable) details of the `RoomEventCache`.
 pub(super) struct RoomEventCacheInner {
+    pub generation: AtomicU64,
     /// The room id for this room.
     room_id: OwnedRoomId,
 
@@ -458,6 +512,7 @@ impl RoomEventCacheInner {
         update_sender: RoomEventCacheUpdateSender,
     ) -> Self {
         Self {
+            generation: AtomicU64::new(0),
             room_id,
             weak_room,
             state,

Modified file: crates/matrix-sdk/src/event_cache/caches/room/pagination.rs+16 additions−3 deletions@@ -154,6 +154,10 @@ impl RoomPagination {
 }
 
 impl PaginatedCache for Arc<RoomEventCacheInner> {
+    fn generation(&self) -> u64 {
+        self.generation.load(std::sync::atomic::Ordering::SeqCst)
+    }
+
     fn status(&self) -> &SharedObservable<SharedPaginationStatus> {
         &self.shared_pagination_status
     }
@@ -308,7 +312,12 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
         events: Vec<Event>,
         timeline_event_diffs: Vec<VectorDiff<Event>>,
         reached_start: bool,
-    ) -> BackPaginationOutcome {
+        generation: u64,
+    ) -> Result<Option<BackPaginationOutcome>> {
+        let _state = self.state.read().await?;
+        if self.generation.load(std::sync::atomic::Ordering::SeqCst) != generation {
+            return Ok(None);
+        }
         if !timeline_event_diffs.is_empty() {
             self.update_sender.send(
                 RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
@@ -319,12 +328,12 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
             );
         }
 
-        BackPaginationOutcome {
+        Ok(Some(BackPaginationOutcome {
             reached_start,
             // This is a backwards pagination. `BackPaginationOutcome` expects events to
             // be in “reverse order”.
             events: events.into_iter().rev().collect(),
-        }
+        }))
     }
 
     async fn conclude_backwards_pagination_from_network(
@@ -332,8 +341,12 @@ impl PaginatedCache for Arc<RoomEventCacheInner> {
         events: Vec<Event>,
         prev_token: Option<String>,
         mut new_token: Option<String>,
+        generation: u64,
     ) -> Result<Option<BackPaginationOutcome>> {
         let mut state = self.state.write().await?;
+        if self.generation.load(std::sync::atomic::Ordering::SeqCst) != generation {
+            return Ok(None);
+        }
 
         // Check that the previous token still exists; otherwise it's a sign that the
         // room's timeline has been cleared.

Added file: crates/matrix-sdk/src/event_cache/caches/room/reseed.md+24 additions@@ -0,0 +1,24 @@
+Replace this room's timeline with independently fetched history and its older
+pagination cursor. The replacement is persisted in one linked chunk update
+transaction before a timeline reset is published.
+
+Pass events in topological order, oldest first. For a backwards
+[`Room::messages`](crate::Room::messages) response, reverse its `chunk` and pass
+its `end` as `prev_batch`. Decryption should happen through the normal room
+messages API before reseeding. Duplicate event IDs within the supplied window
+are retained once. `None` means the start of accessible history has been reached,
+including for an empty window. An empty window with a cursor remains pageable.
+
+This replaces cached history and invalidates loaded thread timelines. It does
+not establish continuity with discarded history. Account credentials, encryption
+state, and pending outgoing messages are not cleared.
+
+Once started, the replacement finishes even if the caller drops its future.
+Failure to persist the replacement leaves the prior room timeline intact.
+Post-processing can fail after the replacement is installed, in which case the
+new timeline is still published and the error is returned. Completion means the
+cache update has been emitted, not that an application mirror has consumed it.
+
+Callers must coordinate live sync and requests that fetched this window. Sync
+updates are serialized against replacement but may append normally afterward.
+In-flight older pagination is fenced against timeline replacement.

Modified file: crates/matrix-sdk/src/event_cache/caches/room/state.rs+25 additions@@ -772,6 +772,31 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
         Ok(diff_updates)
     }
 
+    pub async fn reseed(
+        &mut self,
+        mut events: Vec<Event>,
+        prev_batch: Option<String>,
+    ) -> Result<(), EventCacheError> {
+        let mut event_ids = HashSet::new();
+        events.retain(|event| event.event_id().is_none_or(|id| event_ids.insert(id)));
+        let mut replacement = EventLinkedChunk::new();
+        replacement.reset();
+        replacement.push_live_events(prev_batch.clone().map(|token| Gap { token }), &events);
+        let updates = replacement.store_updates().take();
+        self.send_updates_to_store(updates).await?;
+        let _ = replacement.updates_as_vector_diffs();
+        self.state.room_linked_chunk = replacement;
+        self.state.waited_for_initial_prev_token = true;
+        Ok(())
+    }
+
+    pub async fn clear_loaded_threads(&mut self) -> Result<(), EventCacheError> {
+        for thread in self.state.threads.values_mut() {
+            thread.clear().await?;
+        }
+        Ok(())
+    }
+
     async fn reset_internal(&mut self) -> Result<(), EventCacheError> {
         self.state.room_linked_chunk.reset();
 

Modified file: crates/matrix-sdk/src/event_cache/caches/thread/pagination.rs+3 additions−1 deletions@@ -170,7 +170,8 @@ impl PaginatedCache for ThreadEventCacheWrapper {
         _events: Vec<Event>,
         _timeline_event_diffs: Vec<VectorDiff<Event>>,
         _reached_start: bool,
-    ) -> BackPaginationOutcome {
+        _generation: u64,
+    ) -> Result<Option<BackPaginationOutcome>> {
         unimplemented!("loading from disk for threads is not implemented yet");
     }
 
@@ -179,6 +180,7 @@ impl PaginatedCache for ThreadEventCacheWrapper {
         mut events: Vec<Event>,
         prev_token: Option<String>,
         new_token: Option<String>,
+        _generation: u64,
     ) -> Result<Option<BackPaginationOutcome>> {
         let Some(room) = self.cache.weak_room.get() else {
             // The client is shutting down.

Added file: crates/matrix-sdk/tests/history_reseed.rs+262 additions@@ -0,0 +1,262 @@
+use std::{
+    sync::{Arc, atomic::Ordering},
+    time::Duration,
+};
+
+use eyeball_im::VectorDiff;
+use matrix_sdk::{
+    config::StoreConfig,
+    event_cache::{PaginationStatus, RoomEventCacheUpdate},
+    room::MessagesOptions,
+    ruma::{event_id, room_id},
+    test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate},
+};
+use matrix_sdk_base::{event_cache::store::EventCacheStore, linked_chunk::LinkedChunkId};
+use matrix_sdk_common::cross_process_lock::CrossProcessLockConfig;
+use matrix_sdk_test::{ALICE, event_factory::EventFactory};
+
+#[path = "history_reseed/failing_store.rs"]
+mod failing_store;
+
+#[tokio::test]
+async fn reseed_publishes_one_reset_and_preserves_the_fetched_cursor() {
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+    let room_id = room_id!("!reseed:localhost");
+    client.event_cache().subscribe().unwrap();
+    server.sync_joined_room(&client, room_id).await;
+    let room = client.get_room(room_id).unwrap();
+    let (cache, _handles) = room.event_cache().await.unwrap();
+    let (_, mut updates) = cache.subscribe().await.unwrap();
+    let factory = EventFactory::new().room(room_id).sender(&ALICE);
+    cache
+        .reseed(vec![factory.text_msg("old").into_event()], Some("old-cursor".into()))
+        .await
+        .unwrap();
+    updates.recv().await.unwrap();
+    let recent_id = event_id!("$recent:localhost");
+    let older_id = event_id!("$older:localhost");
+    server
+        .mock_room_messages()
+        .ok(RoomMessagesResponseTemplate::default()
+            .events(vec![
+                factory.text_msg("recent").event_id(recent_id),
+                factory.text_msg("older").event_id(older_id),
+                factory.text_msg("older").event_id(older_id),
+            ])
+            .end_token("fetched-older-cursor"))
+        .mock_once()
+        .mount()
+        .await;
+    let mut history = room.messages(MessagesOptions::backward()).await.unwrap();
+    history.chunk.reverse();
+    cache.reseed(history.chunk, history.end).await.unwrap();
+    let update =
+        tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
+    let RoomEventCacheUpdate::UpdateTimelineEvents(update) = update else {
+        panic!("expected reset")
+    };
+    assert_eq!(update.diffs.len(), 1);
+    let VectorDiff::Reset { values } = &update.diffs[0] else { panic!("expected reset") };
+    assert_eq!(values.len(), 2);
+    assert_eq!(values[0].event_id().as_deref(), Some(older_id));
+    assert_eq!(values[1].event_id().as_deref(), Some(recent_id));
+    assert!(updates.is_empty());
+    assert_eq!(cache.events().await.unwrap().len(), 2);
+    server
+        .mock_room_messages()
+        .match_from("fetched-older-cursor")
+        .ok(RoomMessagesResponseTemplate::default())
+        .mock_once()
+        .mount()
+        .await;
+    assert!(cache.pagination().run_backwards_once(10).await.unwrap().reached_start);
+}
+
+#[tokio::test]
+async fn reseed_empty_history_distinguishes_a_cursor_from_the_start_of_history() {
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+    let room_id = room_id!("!reseed:localhost");
+    client.event_cache().subscribe().unwrap();
+    server.sync_joined_room(&client, room_id).await;
+    let (cache, _handles) = client.get_room(room_id).unwrap().event_cache().await.unwrap();
+    cache.reseed(Vec::new(), None).await.unwrap();
+    assert_eq!(
+        cache.pagination().status().get(),
+        PaginationStatus::Idle { hit_timeline_start: true }
+    );
+    assert!(cache.pagination().run_backwards_once(10).await.unwrap().reached_start);
+    cache.reseed(Vec::new(), Some("empty-page-cursor".into())).await.unwrap();
+    assert_eq!(
+        cache.pagination().status().get(),
+        PaginationStatus::Idle { hit_timeline_start: false }
+    );
+    server
+        .mock_room_messages()
+        .match_from("empty-page-cursor")
+        .ok(RoomMessagesResponseTemplate::default())
+        .mock_once()
+        .mount()
+        .await;
+    assert!(cache.pagination().run_backwards_once(10).await.unwrap().reached_start);
+}
+
+#[tokio::test]
+async fn reseed_fences_inflight_tokenless_pagination() {
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+    let room_id = room_id!("!reseed:localhost");
+    client.event_cache().subscribe().unwrap();
+    server.sync_joined_room(&client, room_id).await;
+    let (cache, _handles) = client.get_room(room_id).unwrap().event_cache().await.unwrap();
+    let factory = EventFactory::new().room(room_id).sender(&ALICE);
+    let stale = factory.text_msg("stale request");
+    server
+        .mock_room_messages()
+        .ok(RoomMessagesResponseTemplate::default()
+            .events(vec![stale])
+            .with_delay(Duration::from_millis(300)))
+        .mock_once()
+        .mount()
+        .await;
+    let pagination = tokio::spawn({
+        let cache = cache.clone();
+        async move { cache.pagination().run_backwards_once(10).await }
+    });
+    tokio::time::timeout(Duration::from_secs(8), async {
+        loop {
+            if server
+                .received_requests()
+                .await
+                .unwrap()
+                .iter()
+                .any(|r| r.url.path().ends_with("/messages"))
+            {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    let recent_id = event_id!("$fresh:localhost");
+    cache
+        .reseed(vec![factory.text_msg("fresh").event_id(recent_id).into_event()], None)
+        .await
+        .unwrap();
+    tokio::time::timeout(Duration::from_secs(3), pagination).await.unwrap().unwrap().unwrap();
+    let events = cache.events().await.unwrap();
+    assert_eq!(events.len(), 1);
+    assert_eq!(events[0].event_id().as_deref(), Some(recent_id));
+    assert_eq!(
+        cache.pagination().status().get(),
+        PaginationStatus::Idle { hit_timeline_start: true }
+    );
+}
+
+#[tokio::test]
+async fn reseed_preserves_timeline_when_store_write_fails() {
+    let server = MatrixMockServer::new().await;
+    let store = Arc::new(failing_store::FailingStore::default());
+    let client = server
+        .client_builder()
+        .on_builder(|builder| {
+            builder.store_config(
+                StoreConfig::new(CrossProcessLockConfig::SingleProcess)
+                    .event_cache_store(store.clone()),
+            )
+        })
+        .build()
+        .await;
+    let room_id = room_id!("!reseed:localhost");
+    client.event_cache().subscribe().unwrap();
+    server.sync_joined_room(&client, room_id).await;
+    let (cache, _handles) = client.get_room(room_id).unwrap().event_cache().await.unwrap();
+    let factory = EventFactory::new().room(room_id).sender(&ALICE);
+    let original_id = event_id!("$original:localhost");
+    cache
+        .reseed(
+            vec![factory.text_msg("original").event_id(original_id).into_event()],
+            Some("original-cursor".into()),
+        )
+        .await
+        .unwrap();
+    let (original_events, updates) = cache.subscribe().await.unwrap();
+    assert_eq!(original_events.len(), 1);
+    assert_eq!(original_events[0].event_id().as_deref(), Some(original_id));
+    let original_chunks = store.inner.load_all_chunks(LinkedChunkId::Room(room_id)).await.unwrap();
+
+    store.fail_writes.store(true, Ordering::SeqCst);
+    let result = cache
+        .reseed(
+            vec![
+                factory
+                    .text_msg("replacement")
+                    .event_id(event_id!("$replacement:localhost"))
+                    .into_event(),
+            ],
+            Some("replacement-cursor".into()),
+        )
+        .await;
+    store.fail_writes.store(false, Ordering::SeqCst);
+
+    assert!(result.is_err());
+    assert!(updates.is_empty());
+    let current_events = cache.events().await.unwrap();
+    assert_eq!(current_events.len(), 1);
+    assert_eq!(current_events[0].event_id().as_deref(), Some(original_id));
+    let current_chunks = store.inner.load_all_chunks(LinkedChunkId::Room(room_id)).await.unwrap();
+    assert_eq!(format!("{current_chunks:?}"), format!("{original_chunks:?}"));
+}
+
+#[cfg(feature = "sqlite")]
+#[tokio::test]
+async fn reseed_persists_the_window_and_cursor_across_client_recreation() {
+    let server = MatrixMockServer::new().await;
+    let directory = tempfile::tempdir().unwrap();
+    let room_id = room_id!("!reseed:localhost");
+    let recent_id = event_id!("$recent:localhost");
+    {
+        let client = server
+            .client_builder()
+            .on_builder(|b| b.sqlite_store(directory.path(), None))
+            .build()
+            .await;
+        client.event_cache().subscribe().unwrap();
+        server.sync_joined_room(&client, room_id).await;
+        let (cache, _handles) = client.get_room(room_id).unwrap().event_cache().await.unwrap();
+        cache
+            .reseed(
+                vec![
+                    EventFactory::new()
+                        .room(room_id)
+                        .sender(&ALICE)
+                        .text_msg("recent")
+                        .event_id(recent_id)
+                        .into_event(),
+                ],
+                Some("persisted-cursor".into()),
+            )
+            .await
+            .unwrap();
+    }
+    let client = server
+        .client_builder()
+        .on_builder(|b| b.sqlite_store(directory.path(), None))
+        .build()
+        .await;
+    client.event_cache().subscribe().unwrap();
+    let (cache, _handles) = client.get_room(room_id).unwrap().event_cache().await.unwrap();
+    let (events, _subscriber) = cache.subscribe().await.unwrap();
+    assert_eq!(events.len(), 1);
+    assert_eq!(events[0].event_id().as_deref(), Some(recent_id));
+    server
+        .mock_room_messages()
+        .match_from("persisted-cursor")
+        .ok(RoomMessagesResponseTemplate::default())
+        .mock_once()
+        .mount()
+        .await;
+    assert!(cache.pagination().run_backwards_once(10).await.unwrap().reached_start);
+}

Added file: crates/matrix-sdk/tests/history_reseed/failing_store.rs+141 additions@@ -0,0 +1,141 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use async_trait::async_trait;
+use matrix_sdk_base::{
+    event_cache::{
+        Event, Gap,
+        store::{EventCacheStore, EventCacheStoreError, MemoryStore},
+    },
+    linked_chunk::{
+        ChunkIdentifier, ChunkIdentifierGenerator, ChunkMetadata, LinkedChunkId, Position,
+        RawChunk, Update,
+    },
+};
+use matrix_sdk_common::cross_process_lock::CrossProcessLockGeneration;
+use ruma::{EventId, OwnedEventId, RoomId, events::relation::RelationType};
+
+#[derive(Debug, Default)]
+pub struct FailingStore {
+    pub inner: MemoryStore,
+    pub fail_writes: AtomicBool,
+}
+
+#[cfg_attr(target_family = "wasm", async_trait(?Send))]
+#[cfg_attr(not(target_family = "wasm"), async_trait)]
+impl EventCacheStore for FailingStore {
+    type Error = EventCacheStoreError;
+
+    async fn try_take_leased_lock(
+        &self,
+        lease_duration_ms: u32,
+        key: &str,
+        holder: &str,
+    ) -> Result<Option<CrossProcessLockGeneration>, Self::Error> {
+        self.inner.try_take_leased_lock(lease_duration_ms, key, holder).await
+    }
+
+    async fn handle_linked_chunk_updates(
+        &self,
+        linked_chunk_id: LinkedChunkId<'_>,
+        updates: Vec<Update<Event, Gap>>,
+    ) -> Result<(), Self::Error> {
+        if self.fail_writes.load(Ordering::SeqCst) {
+            return Err(EventCacheStoreError::InvalidData {
+                details: "injected write failure".into(),
+            });
+        }
+
+        self.inner.handle_linked_chunk_updates(linked_chunk_id, updates).await
+    }
+
+    async fn remove_room(&self, room_id: &RoomId) -> Result<(), Self::Error> {
+        self.inner.remove_room(room_id).await
+    }
+
+    async fn load_all_chunks(
+        &self,
+        linked_chunk_id: LinkedChunkId<'_>,
+    ) -> Result<Vec<RawChunk<Event, Gap>>, Self::Error> {
+        self.inner.load_all_chunks(linked_chunk_id).await
+    }
+
+    async fn load_all_chunks_metadata(
+        &self,
+        linked_chunk_id: LinkedChunkId<'_>,
+    ) -> Result<Vec<ChunkMetadata>, Self::Error> {
+        self.inner.load_all_chunks_metadata(linked_chunk_id).await
+    }
+
+    async fn load_last_chunk(
+        &self,
+        linked_chunk_id: LinkedChunkId<'_>,
+    ) -> Result<(Option<RawChunk<Event, Gap>>, ChunkIdentifierGenerator), Self::Error> {
+        self.inner.load_last_chunk(linked_chunk_id).await
+    }
+
+    async fn load_previous_chunk(
+        &self,
+        linked_chunk_id: LinkedChunkId<'_>,
+        before_chunk_identifier: ChunkIdentifier,
+    ) -> Result<Option<RawChunk<Event, Gap>>, Self::Error> {
+        self.inner.load_previous_chunk(linked_chunk_id, before_chunk_identifier).await
+    }
+
+    async fn clear_all_linked_chunks(&self) -> Result<(), Self::Error> {
+        self.inner.clear_all_linked_chunks().await
+    }
+
+    async fn filter_duplicated_events(
+        &self,
+        linked_chunk_id: LinkedChunkId<'_>,
+        events: Vec<OwnedEventId>,
+    ) -> Result<Vec<(OwnedEventId, Position)>, Self::Error> {
+        self.inner.filter_duplicated_events(linked_chunk_id, events).await
+    }
+
+    async fn find_event(
+        &self,
+        room_id: &RoomId,
+        event_id: &EventId,
+    ) -> Result<Option<Event>, Self::Error> {
+        self.inner.find_event(room_id, event_id).await
+    }
+
+    async fn find_event_relations(
+        &self,
+        room_id: &RoomId,
+        event_id: &EventId,
+        filter: Option<&[RelationType]>,
+    ) -> Result<Vec<(Event, Option<Position>)>, Self::Error> {
+        self.inner.find_event_relations(room_id, event_id, filter).await
+    }
+
+    async fn get_room_events(
+        &self,
+        room_id: &RoomId,
+        event_type: Option<&str>,
+        session_id: Option<&str>,
+    ) -> Result<Vec<Event>, Self::Error> {
+        self.inner.get_room_events(room_id, event_type, session_id).await
+    }
+
+    async fn save_event(&self, room_id: &RoomId, event: Event) -> Result<(), Self::Error> {
+        self.inner.save_event(room_id, event).await
+    }
+
+    async fn close(&self) -> Result<(), Self::Error> {
+        self.inner.close().await
+    }
+
+    async fn reopen(&self) -> Result<(), Self::Error> {
+        self.inner.reopen().await
+    }
+
+    async fn optimize(&self) -> Result<(), Self::Error> {
+        self.inner.optimize().await
+    }
+
+    async fn get_size(&self) -> Result<Option<usize>, Self::Error> {
+        self.inner.get_size().await
+    }
+}

matrix-rust-sdk

Commit ef3c294b31cc

Commit `ef3c294b31cc`