matrix-rust-sdk

public

Star0

Fork 0

Watch0

19 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `4fae3b516f05`

tarball

atagen <boss@atagen.co>2026-10-25 08:58 UTC

13 files changed+437-1010

event-cache: drop RoomEventCache::reseed

macltok moved to fill_history_gap_once and recent loads, so reseed was about 500 lines of fork surface with no caller. The generation fence stays for clear and commit_history_candidate, and the clear tests move to their own target with the history tests seeding through sync.

Modified file: crates/matrix-sdk-ui/tests/integration/timeline/mod.rs−1

@@ -56,7 +56,6 @@ mod queue;

5656 mod reactions;

5757 mod read_receipts;

5858 mod redecryption;

59-mod reseed;

6059 mod segmented_history;

6160 mod replies;

6261 mod subscribe;

Deleted file: crates/matrix-sdk-ui/tests/integration/timeline/reseed.rs−109 deletions@@ -1,109 +0,0 @@
-use std::time::Duration;
-
-use assert_matches::assert_matches;
-use futures_util::StreamExt;
-use matrix_sdk::test_utils::mocks::MatrixMockServer;
-use matrix_sdk_test::{ALICE, JoinedRoomBuilder, async_test, event_factory::EventFactory};
-use matrix_sdk_ui::timeline::{EventSendState, RoomExt};
-use ruma::{event_id, events::room::message::RoomMessageEventContent, room_id};
-use tokio::time::timeout;
-
-#[async_test]
-async fn test_timeline_survives_room_cache_reseed() {
-    let server = MatrixMockServer::new().await;
-    let client = server.client_builder().build().await;
-    client.send_queue().set_enabled(false).await;
-
-    server.mock_room_state_encryption().plain().mount().await;
-
-    let room_id = room_id!("!a98sd12bjh:example.org");
-    let room = server.sync_joined_room(&client, room_id).await;
-    let timeline = room.timeline().await.unwrap();
-    let (_, mut timeline_stream) = timeline.subscribe().await;
-    let event_factory = EventFactory::new().room(room_id);
-
-    let old_event_id = event_id!("$before_reseed");
-    server
-        .sync_room(
-            &client,
-            JoinedRoomBuilder::new(room_id).add_timeline_event(
-                event_factory.text_msg("old history").sender(&ALICE).event_id(old_event_id),
-            ),
-        )
-        .await;
-
-    timeout(Duration::from_secs(5), async {
-        loop {
-            let items = timeline.items().await;
-            if items.iter().any(|item| {
-                item.as_event().is_some_and(|event| event.event_id() == Some(old_event_id))
-            }) {
-                break;
-            }
-            assert!(timeline_stream.next().await.is_some());
-        }
-    })
-    .await
-    .unwrap();
-
-    timeline.send(RoomMessageEventContent::text_plain("pending local echo").into()).await.unwrap();
-
-    timeout(Duration::from_secs(5), async {
-        loop {
-            let items = timeline.items().await;
-            if items.iter().any(|item| {
-                item.as_event().is_some_and(|event| {
-                    event
-                        .content()
-                        .as_message()
-                        .is_some_and(|message| message.body() == "pending local echo")
-                })
-            }) {
-                break;
-            }
-            assert!(timeline_stream.next().await.is_some());
-        }
-    })
-    .await
-    .unwrap();
-
-    let new_event_id = event_id!("$after_reseed");
-    let new_event =
-        event_factory.text_msg("recent history").sender(&ALICE).event_id(new_event_id).into_event();
-    let (room_event_cache, _drop_handles) = room.event_cache().await.unwrap();
-    room_event_cache
-        .reseed(vec![new_event], Some("older".into()))
-        .await
-        .unwrap();
-
-    timeout(Duration::from_secs(5), async {
-        loop {
-            let items = timeline.items().await;
-            if items.iter().any(|item| {
-                item.as_event().is_some_and(|event| event.event_id() == Some(new_event_id))
-            }) {
-                break;
-            }
-            assert!(timeline_stream.next().await.is_some());
-        }
-    })
-    .await
-    .unwrap();
-
-    let items = timeline.items().await;
-    assert!(!items.iter().any(|item| {
-        item.as_event().is_some_and(|event| event.event_id() == Some(old_event_id))
-    }));
-
-    let local_echo = items
-        .iter()
-        .filter_map(|item| item.as_event())
-        .find(|event| {
-            event
-                .content()
-                .as_message()
-                .is_some_and(|message| message.body() == "pending local echo")
-        })
-        .unwrap();
-    assert_matches!(local_echo.send_state(), Some(EventSendState::NotSentYet { .. }));
-}

Modified file: crates/matrix-sdk-ui/tests/integration/timeline/segmented_history.rs+8 additions−7 deletions@@ -2,7 +2,7 @@ use std::time::Duration;
 
 use futures_util::StreamExt;
 use matrix_sdk::test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate};
-use matrix_sdk_test::{ALICE, async_test, event_factory::EventFactory};
+use matrix_sdk_test::{ALICE, JoinedRoomBuilder, async_test, event_factory::EventFactory};
 use matrix_sdk_ui::timeline::{RoomExt, VirtualTimelineItem};
 use ruma::{event_id, events::room::message::RoomMessageEventContent, room_id};
 use tokio::time::timeout;
@@ -22,13 +22,14 @@ async fn test_timeline_fills_segmented_history_without_replacing_retained_items(
     let old_event_id = event_id!("$old");
     let hidden_event_id = event_id!("$hidden");
     let new_event_id = event_id!("$new");
-    cache
-        .reseed(
-            vec![event_factory.text_msg("old").event_id(old_event_id).into_event()],
-            Some("old-cursor".into()),
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(event_factory.text_msg("old").event_id(old_event_id))
+                .set_timeline_prev_batch("old-cursor"),
         )
-        .await
-        .unwrap();
+        .await;
     let gap = cache
         .append_history_section(
             vec![

Modified file: crates/matrix-sdk/Cargo.toml+1−1

@@ -210,7 +210,7 @@ name = "response_commit"

210210 required-features = ["testing", "e2e-encryption"]

211211

212212 [[test]]

213-name = "history_reseed"

213+name = "room_clear"

214214 required-features = ["testing"]

215215

216216 [[test]]

Modified file: crates/matrix-sdk/src/event_cache/caches/room/mod.rs−35 deletions@@ -384,41 +384,6 @@ 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 events =
-                state.room_linked_chunk().events().map(|(_, event)| event.clone()).collect();
-            let result = async {
-                state.clear_loaded_views().await?;
-                state
-                    .post_process_new_events(events, PostProcessingOrigin::Backpagination, None)
-                    .await
-            }
-            .await;
-            let items =
-                state.room_linked_chunk().events().map(|(_, event)| event.clone()).collect();
-            let _ = state.room_linked_chunk_mut().updates_as_vector_diffs();
-            cache.inner.update_sender.send(
-                RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
-                    diffs: vec![eyeball_im::VectorDiff::Reset { values: items }],
-                    history_gaps: Some(history::gap_positions(state.room_linked_chunk())),
-                    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
-        .map_err(|error| EventCacheError::TaskInterrupted(error.to_string()))?
-    }
-
     /// Return a reference to the state.
     pub(in super::super) fn state(&self) -> &LockedRoomEventCacheState {
         &self.inner.state

Deleted file: crates/matrix-sdk/src/event_cache/caches/room/reseed.md−25 deletions@@ -1,25 +0,0 @@
-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 thread, pinned, and event-focused
-timelines, including persisted views that have not been loaded. 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−22 deletions@@ -781,28 +781,6 @@ 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, history: None }),
-            &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;
-        self.reset_pagination_status(prev_batch.is_none());
-        Ok(())
-    }
-
     pub(super) async fn history_candidate(&self) -> Result<EventLinkedChunk, EventCacheError> {
         let linked_chunk_id = LinkedChunkId::Room(&self.state.room_id);
         let (last, generator) = self.store.load_last_chunk(linked_chunk_id).await?;

Deleted file: crates/matrix-sdk/tests/history_reseed.rs−598 deletions@@ -1,598 +0,0 @@
-use std::{
-    sync::{Arc, atomic::Ordering},
-    time::Duration,
-};
-
-use eyeball_im::VectorDiff;
-use matrix_sdk::{
-    ThreadingSupport,
-    config::StoreConfig,
-    event_cache::{EventFocusThreadMode, PaginationStatus, RoomEventCacheUpdate},
-    room::MessagesOptions,
-    ruma::{event_id, room_id},
-    test_utils::mocks::{
-        MatrixMockServer, RoomContextResponseTemplate, RoomMessagesResponseTemplate,
-        RoomRelationsResponseTemplate,
-    },
-};
-use matrix_sdk_base::{event_cache::store::EventCacheStore, linked_chunk::LinkedChunkId};
-use matrix_sdk_common::cross_process_lock::CrossProcessLockConfig;
-use matrix_sdk_test::{ALICE, JoinedRoomBuilder, 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:?}"));
-}
-
-#[tokio::test]
-async fn reseed_redaction_does_not_leak_stale_diff_into_next_sync() {
-    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);
-    let redacted_id = event_id!("$redacted:localhost");
-    let redaction_id = event_id!("$redaction:localhost");
-
-    cache
-        .reseed(
-            vec![
-                factory.text_msg("secret").event_id(redacted_id).into_event(),
-                factory.redaction(&redacted_id).event_id(redaction_id).into_event(),
-            ],
-            None,
-        )
-        .await
-        .unwrap();
-    let reset =
-        tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
-    let RoomEventCacheUpdate::UpdateTimelineEvents(reset) = reset else {
-        panic!("expected timeline reset")
-    };
-    let VectorDiff::Reset { values } = &reset.diffs[0] else { panic!("expected reset diff") };
-    assert_eq!(values.len(), 2);
-    assert!(values[0].raw().deserialize().unwrap().is_redacted());
-
-    let new_event_id = event_id!("$new-live:localhost");
-    server
-        .sync_room(
-            &client,
-            JoinedRoomBuilder::new(room_id).add_timeline_event(
-                factory.text_msg("live").event_id(new_event_id).into_raw_sync(),
-            ),
-        )
-        .await;
-    let live_update =
-        tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
-    let RoomEventCacheUpdate::UpdateTimelineEvents(live_update) = live_update else {
-        panic!("expected timeline update")
-    };
-    assert!(live_update.diffs.iter().all(|diff| !matches!(diff, VectorDiff::Set { .. })));
-    assert!(live_update.diffs.iter().all(|diff| match diff {
-        VectorDiff::Append { values } => {
-            values.iter().all(|event| event.event_id().as_deref() == Some(new_event_id))
-        }
-        VectorDiff::PushBack { value } => value.event_id().as_deref() == Some(new_event_id),
-        _ => true,
-    }));
-    assert_eq!(cache.events().await.unwrap().len(), 3);
-}
-
-#[tokio::test]
-async fn reseed_retries_thread_pagination_without_stale_response() {
-    use wiremock::ResponseTemplate;
-
-    let server = MatrixMockServer::new().await;
-    let client = server
-        .client_builder()
-        .on_builder(|builder| {
-            builder.with_threading_support(ThreadingSupport::Enabled { with_subscriptions: false })
-        })
-        .build()
-        .await;
-    let room_id = room_id!("!reseed:localhost");
-    client.event_cache().subscribe().unwrap();
-    let thread_root_id = event_id!("$thread-root:localhost");
-    let initial_reply_id = event_id!("$initial-reply:localhost");
-    let factory = EventFactory::new().room(room_id).sender(&ALICE);
-    let room = server
-        .sync_room(
-            &client,
-            JoinedRoomBuilder::new(room_id)
-                .set_timeline_prev_batch("initial-thread-cursor")
-                .add_timeline_event(
-                    factory
-                        .text_msg("initial reply")
-                        .in_thread(thread_root_id, thread_root_id)
-                        .event_id(initial_reply_id),
-                ),
-        )
-        .await;
-    let (cache, _handles) = room.event_cache().await.unwrap();
-    let (initial_events, _thread_updates) =
-        cache.subscribe_to_thread(thread_root_id.to_owned()).await.unwrap();
-    assert_eq!(initial_events.len(), 1);
-    assert_eq!(initial_events[0].event_id().as_deref(), Some(initial_reply_id));
-
-    let stale_reply_id = event_id!("$stale-reply:localhost");
-    let fresh_reply_id = event_id!("$fresh-reply:localhost");
-    server
-        .mock_room_relations()
-        .match_target_event(thread_root_id.to_owned())
-        .ok(RoomRelationsResponseTemplate::default().events(vec![
-            factory
-                .text_msg("stale reply")
-                .in_thread(thread_root_id, thread_root_id)
-                .event_id(stale_reply_id),
-        ]))
-        .mock_once()
-        .mount()
-        .await;
-    let root_event = factory.text_msg("thread root").event_id(thread_root_id).into_event();
-    server
-        .mock_room_event()
-        .room(room_id)
-        .ok_with_template(
-            ResponseTemplate::new(200)
-                .set_body_json(root_event.into_raw().json())
-                .set_delay(Duration::from_millis(300)),
-        )
-        .mock_once()
-        .mount()
-        .await;
-
-    let pagination = tokio::spawn({
-        let cache = cache.clone();
-        let thread_root_id = thread_root_id.to_owned();
-        async move {
-            cache.thread_pagination(thread_root_id).await.unwrap().run_backwards_once(10).await
-        }
-    });
-    tokio::time::timeout(Duration::from_secs(8), async {
-        loop {
-            if server
-                .received_requests()
-                .await
-                .unwrap()
-                .iter()
-                .any(|request| request.url.path().contains("/event/"))
-            {
-                break;
-            }
-            tokio::task::yield_now().await;
-        }
-    })
-    .await
-    .unwrap();
-    server
-        .mock_room_relations()
-        .match_target_event(thread_root_id.to_owned())
-        .ok(RoomRelationsResponseTemplate::default()
-            .events(vec![
-                factory
-                    .text_msg("fresh reply")
-                    .in_thread(thread_root_id, thread_root_id)
-                    .event_id(fresh_reply_id),
-            ])
-            .next_batch("fresh-cursor"))
-        .mock_once()
-        .mount()
-        .await;
-    cache.reseed(Vec::new(), None).await.unwrap();
-
-    let outcome =
-        tokio::time::timeout(Duration::from_secs(8), pagination).await.unwrap().unwrap().unwrap();
-    assert!(outcome.events.iter().all(|event| event.event_id().as_deref() != Some(stale_reply_id)));
-    assert!(outcome.events.iter().any(|event| event.event_id().as_deref() == Some(fresh_reply_id)));
-    let (thread_events, _thread_updates) =
-        cache.subscribe_to_thread(thread_root_id.to_owned()).await.unwrap();
-    assert!(thread_events.iter().all(|event| event.event_id().as_deref() != Some(stale_reply_id)));
-    assert!(thread_events.iter().any(|event| event.event_id().as_deref() == Some(fresh_reply_id)));
-}
-
-#[tokio::test]
-async fn clear_invalidates_focused_cache_clones_and_inflight_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 room = client.get_room(room_id).unwrap();
-    let (cache, _handles) = room.event_cache().await.unwrap();
-    let factory = EventFactory::new().room(room_id).sender(&ALICE);
-    let focused_id = event_id!("$focused:localhost");
-    server
-        .mock_room_event_context()
-        .ok(RoomContextResponseTemplate::new(
-            factory.text_msg("focused").event_id(focused_id).into_event(),
-        )
-        .start("old-focused-cursor"))
-        .mock_once()
-        .mount()
-        .await;
-    let focused = cache
-        .get_or_create_event_focused_cache(
-            focused_id.to_owned(),
-            10,
-            EventFocusThreadMode::Automatic,
-        )
-        .await
-        .unwrap();
-    let focused_clone = focused.clone();
-    let (initial_events, mut updates) = focused.subscribe().await;
-    assert_eq!(initial_events.len(), 1);
-    assert_eq!(initial_events[0].event_id().as_deref(), Some(focused_id));
-
-    let stale_event_id = event_id!("$stale-focused:localhost");
-    server
-        .mock_room_messages()
-        .match_from("old-focused-cursor")
-        .ok(RoomMessagesResponseTemplate::default()
-            .events(vec![factory.text_msg("stale").event_id(stale_event_id)])
-            .end_token("stale-next-cursor")
-            .with_delay(Duration::from_millis(300)))
-        .mock_once()
-        .mount()
-        .await;
-    let pagination = tokio::spawn({
-        let focused = focused.clone();
-        async move { focused.paginate_backwards(10).await }
-    });
-    tokio::time::timeout(Duration::from_secs(8), async {
-        loop {
-            if server
-                .received_requests()
-                .await
-                .unwrap()
-                .iter()
-                .any(|request| request.url.path().ends_with("/messages"))
-            {
-                break;
-            }
-            tokio::task::yield_now().await;
-        }
-    })
-    .await
-    .unwrap();
-
-    tokio::time::timeout(Duration::from_secs(2), cache.clear()).await.unwrap().unwrap();
-    let clear =
-        tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
-    assert!(matches!(clear.diffs.as_slice(), [VectorDiff::Clear]));
-    tokio::time::timeout(Duration::from_secs(3), pagination).await.unwrap().unwrap().unwrap();
-    let (cloned_events, _cloned_updates) = focused_clone.subscribe().await;
-    assert!(cloned_events.is_empty());
-    let result = focused_clone.paginate_backwards(10).await.unwrap();
-    assert!(result.hit_end_of_timeline);
-    let (events_after_pagination, _updates_after_pagination) = focused_clone.subscribe().await;
-    assert!(events_after_pagination.is_empty());
-}
-
-#[tokio::test]
-async fn clear_empties_subscribed_thread_cache() {
-    let server = MatrixMockServer::new().await;
-    let client = server
-        .client_builder()
-        .on_builder(|builder| {
-            builder.with_threading_support(ThreadingSupport::Enabled { with_subscriptions: false })
-        })
-        .build()
-        .await;
-    let room_id = room_id!("!reseed:localhost");
-    client.event_cache().subscribe().unwrap();
-    let thread_root_id = event_id!("$thread-root:localhost");
-    let thread_reply_id = event_id!("$thread-reply:localhost");
-    let factory = EventFactory::new().room(room_id).sender(&ALICE);
-    let room = server
-        .sync_room(
-            &client,
-            JoinedRoomBuilder::new(room_id).add_timeline_event(
-                factory
-                    .text_msg("thread reply")
-                    .in_thread(thread_root_id, thread_root_id)
-                    .event_id(thread_reply_id),
-            ),
-        )
-        .await;
-    let (cache, _handles) = room.event_cache().await.unwrap();
-    let (initial_events, mut updates) =
-        cache.subscribe_to_thread(thread_root_id.to_owned()).await.unwrap();
-    assert_eq!(initial_events.len(), 1);
-    assert_eq!(initial_events[0].event_id().as_deref(), Some(thread_reply_id));
-
-    cache.clear().await.unwrap();
-    let clear =
-        tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
-    assert!(matches!(clear.diffs.as_slice(), [VectorDiff::Clear]));
-    let (events_after_clear, _updates_after_clear) =
-        cache.subscribe_to_thread(thread_root_id.to_owned()).await.unwrap();
-    assert!(events_after_clear.is_empty());
-}
-
-#[tokio::test]
-async fn clear_empties_loaded_pinned_cache() {
-    let server = MatrixMockServer::new().await;
-    let room_id = room_id!("!reseed:localhost");
-    let pinned_event_id = event_id!("$pinned:localhost");
-    let factory = EventFactory::new().room(room_id).sender(&ALICE);
-    let pinned_event = factory.text_msg("pinned").event_id(pinned_event_id).into_event();
-    server.mock_room_event().match_event_id().ok(pinned_event).mock_once().mount().await;
-    let client = server.client_builder().build().await;
-    client.event_cache().subscribe().unwrap();
-    let room = server
-        .sync_room(
-            &client,
-            JoinedRoomBuilder::new(room_id).add_state_bulk(vec![
-                factory.room_pinned_events(vec![pinned_event_id.to_owned()]).into(),
-            ]),
-        )
-        .await;
-    let (cache, _handles) = room.event_cache().await.unwrap();
-    let (initial_events, mut updates) = cache.subscribe_to_pinned_events().await.unwrap();
-    let mut visible: imbl::Vector<_> = initial_events.into();
-    tokio::time::timeout(Duration::from_secs(3), async {
-        while visible.is_empty() {
-            for diff in updates.recv().await.unwrap().diffs {
-                diff.apply(&mut visible);
-            }
-        }
-    })
-    .await
-    .unwrap();
-    assert_eq!(visible.len(), 1);
-    assert_eq!(visible[0].event_id().as_deref(), Some(pinned_event_id));
-
-    cache.clear().await.unwrap();
-    let clear =
-        tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
-    assert!(matches!(clear.diffs.as_slice(), [VectorDiff::Clear]));
-    for diff in clear.diffs {
-        diff.apply(&mut visible);
-    }
-    assert!(visible.is_empty());
-}
-
-#[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);
-}

Deleted file: crates/matrix-sdk/tests/history_reseed/failing_store.rs−141 deletions@@ -1,141 +0,0 @@
-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
-    }
-}

Modified file: crates/matrix-sdk/tests/recent_history.rs+63 additions−36 deletions@@ -54,10 +54,13 @@ async fn test_recent_history_overlapping_a_limited_sync_drops_its_stale_gap() {
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = event_factory(room_id);
-    cache
-        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
-        .await
-        .unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),
+        )
+        .await;
     server
         .mock_room_messages()
         .match_limit(50)
@@ -91,10 +94,13 @@ async fn test_recent_history_ending_before_a_limited_sync_keeps_its_gap_after_th
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = event_factory(room_id);
-    cache
-        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
-        .await
-        .unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),
+        )
+        .await;
     server
         .mock_room_messages()
         .match_limit(50)
@@ -129,10 +135,13 @@ async fn test_recent_history_is_bounded_and_keeps_the_retained_window() {
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = event_factory(room_id);
-    cache
-        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
-        .await
-        .unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),
+        )
+        .await;
 
     server
         .mock_room_messages()
@@ -202,10 +211,13 @@ async fn test_recent_history_joins_at_the_retained_boundary() {
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = event_factory(room_id);
-    cache
-        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
-        .await
-        .unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),
+        )
+        .await;
     server
         .mock_room_messages()
         .match_limit(50)
@@ -242,10 +254,13 @@ async fn test_recent_history_keeps_sync_arrivals_during_a_fetch() {
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = event_factory(room_id);
-    cache
-        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
-        .await
-        .unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),
+        )
+        .await;
     server
         .mock_room_messages()
         .match_limit(50)
@@ -288,10 +303,13 @@ async fn test_recent_history_continues_empty_pages_and_pauses_on_a_cursor_cycle(
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = event_factory(room_id);
-    cache
-        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
-        .await
-        .unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),
+        )
+        .await;
     server
         .mock_room_messages()
         .match_limit(50)
@@ -343,10 +361,13 @@ async fn test_concurrent_recent_history_calls_share_the_request_and_gap() {
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = event_factory(room_id);
-    cache
-        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
-        .await
-        .unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),
+        )
+        .await;
     server
         .mock_room_messages()
         .match_limit(50)
@@ -377,10 +398,13 @@ async fn test_cancelling_one_recent_history_waiter_keeps_the_shared_request_aliv
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = event_factory(room_id);
-    cache
-        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
-        .await
-        .unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),
+        )
+        .await;
     server
         .mock_room_messages()
         .match_limit(50)
@@ -426,10 +450,13 @@ async fn test_cancelling_all_recent_history_waiters_allows_a_fresh_retry() {
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = event_factory(room_id);
-    cache
-        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
-        .await
-        .unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),
+        )
+        .await;
     server
         .mock_room_messages()
         .match_limit(50)

Added file: crates/matrix-sdk/tests/room_clear.rs+174 additions@@ -0,0 +1,174 @@
+use std::time::Duration;
+
+use eyeball_im::VectorDiff;
+use matrix_sdk::{
+    ThreadingSupport,
+    event_cache::EventFocusThreadMode,
+    ruma::{event_id, room_id},
+    test_utils::mocks::{
+        MatrixMockServer, RoomContextResponseTemplate, RoomMessagesResponseTemplate,
+    },
+};
+use matrix_sdk_test::{ALICE, JoinedRoomBuilder, event_factory::EventFactory};
+
+#[tokio::test]
+async fn clear_invalidates_focused_cache_clones_and_inflight_pagination() {
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+    let room_id = room_id!("!clear: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 factory = EventFactory::new().room(room_id).sender(&ALICE);
+    let focused_id = event_id!("$focused:localhost");
+    server
+        .mock_room_event_context()
+        .ok(RoomContextResponseTemplate::new(
+            factory.text_msg("focused").event_id(focused_id).into_event(),
+        )
+        .start("old-focused-cursor"))
+        .mock_once()
+        .mount()
+        .await;
+    let focused = cache
+        .get_or_create_event_focused_cache(
+            focused_id.to_owned(),
+            10,
+            EventFocusThreadMode::Automatic,
+        )
+        .await
+        .unwrap();
+    let focused_clone = focused.clone();
+    let (initial_events, mut updates) = focused.subscribe().await;
+    assert_eq!(initial_events.len(), 1);
+    assert_eq!(initial_events[0].event_id().as_deref(), Some(focused_id));
+
+    let stale_event_id = event_id!("$stale-focused:localhost");
+    server
+        .mock_room_messages()
+        .match_from("old-focused-cursor")
+        .ok(RoomMessagesResponseTemplate::default()
+            .events(vec![factory.text_msg("stale").event_id(stale_event_id)])
+            .end_token("stale-next-cursor")
+            .with_delay(Duration::from_millis(300)))
+        .mock_once()
+        .mount()
+        .await;
+    let pagination = tokio::spawn({
+        let focused = focused.clone();
+        async move { focused.paginate_backwards(10).await }
+    });
+    tokio::time::timeout(Duration::from_secs(8), async {
+        loop {
+            if server
+                .received_requests()
+                .await
+                .unwrap()
+                .iter()
+                .any(|request| request.url.path().ends_with("/messages"))
+            {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+
+    tokio::time::timeout(Duration::from_secs(2), cache.clear()).await.unwrap().unwrap();
+    let clear =
+        tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
+    assert!(matches!(clear.diffs.as_slice(), [VectorDiff::Clear]));
+    tokio::time::timeout(Duration::from_secs(3), pagination).await.unwrap().unwrap().unwrap();
+    let (cloned_events, _cloned_updates) = focused_clone.subscribe().await;
+    assert!(cloned_events.is_empty());
+    let result = focused_clone.paginate_backwards(10).await.unwrap();
+    assert!(result.hit_end_of_timeline);
+    let (events_after_pagination, _updates_after_pagination) = focused_clone.subscribe().await;
+    assert!(events_after_pagination.is_empty());
+}
+
+#[tokio::test]
+async fn clear_empties_subscribed_thread_cache() {
+    let server = MatrixMockServer::new().await;
+    let client = server
+        .client_builder()
+        .on_builder(|builder| {
+            builder.with_threading_support(ThreadingSupport::Enabled { with_subscriptions: false })
+        })
+        .build()
+        .await;
+    let room_id = room_id!("!clear:localhost");
+    client.event_cache().subscribe().unwrap();
+    let thread_root_id = event_id!("$thread-root:localhost");
+    let thread_reply_id = event_id!("$thread-reply:localhost");
+    let factory = EventFactory::new().room(room_id).sender(&ALICE);
+    let room = server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id).add_timeline_event(
+                factory
+                    .text_msg("thread reply")
+                    .in_thread(thread_root_id, thread_root_id)
+                    .event_id(thread_reply_id),
+            ),
+        )
+        .await;
+    let (cache, _handles) = room.event_cache().await.unwrap();
+    let (initial_events, mut updates) =
+        cache.subscribe_to_thread(thread_root_id.to_owned()).await.unwrap();
+    assert_eq!(initial_events.len(), 1);
+    assert_eq!(initial_events[0].event_id().as_deref(), Some(thread_reply_id));
+
+    cache.clear().await.unwrap();
+    let clear =
+        tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
+    assert!(matches!(clear.diffs.as_slice(), [VectorDiff::Clear]));
+    let (events_after_clear, _updates_after_clear) =
+        cache.subscribe_to_thread(thread_root_id.to_owned()).await.unwrap();
+    assert!(events_after_clear.is_empty());
+}
+
+#[tokio::test]
+async fn clear_empties_loaded_pinned_cache() {
+    let server = MatrixMockServer::new().await;
+    let room_id = room_id!("!clear:localhost");
+    let pinned_event_id = event_id!("$pinned:localhost");
+    let factory = EventFactory::new().room(room_id).sender(&ALICE);
+    let pinned_event = factory.text_msg("pinned").event_id(pinned_event_id).into_event();
+    server.mock_room_event().match_event_id().ok(pinned_event).mock_once().mount().await;
+    let client = server.client_builder().build().await;
+    client.event_cache().subscribe().unwrap();
+    let room = server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id).add_state_bulk(vec![
+                factory.room_pinned_events(vec![pinned_event_id.to_owned()]).into(),
+            ]),
+        )
+        .await;
+    let (cache, _handles) = room.event_cache().await.unwrap();
+    let (initial_events, mut updates) = cache.subscribe_to_pinned_events().await.unwrap();
+    let mut visible: imbl::Vector<_> = initial_events.into();
+    tokio::time::timeout(Duration::from_secs(3), async {
+        while visible.is_empty() {
+            for diff in updates.recv().await.unwrap().diffs {
+                diff.apply(&mut visible);
+            }
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(visible.len(), 1);
+    assert_eq!(visible[0].event_id().as_deref(), Some(pinned_event_id));
+
+    cache.clear().await.unwrap();
+    let clear =
+        tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
+    assert!(matches!(clear.diffs.as_slice(), [VectorDiff::Clear]));
+    for diff in clear.diffs {
+        diff.apply(&mut visible);
+    }
+    assert!(visible.is_empty());
+}

Modified file: crates/matrix-sdk/tests/segmented_history.rs+50 additions−35 deletions@@ -13,7 +13,7 @@ use matrix_sdk_common::cross_process_lock::CrossProcessLockConfig;
 use matrix_sdk_test::{JoinedRoomBuilder, async_test, event_factory::EventFactory};
 use ruma::{event_id, room_id, user_id};
 
-#[path = "history_reseed/failing_store.rs"]
+#[path = "support/failing_store.rs"]
 mod failing_store;
 
 #[async_test]
@@ -34,13 +34,14 @@ async fn test_segmented_history_resumes_after_restart_and_joins_only_at_boundary
         client.event_cache().subscribe().unwrap();
         let room = server.sync_joined_room(&client, room_id).await;
         let (cache, _handles) = room.event_cache().await.unwrap();
-        cache
-            .reseed(
-                vec![f.text_msg("retained").event_id(event_id!("$old")).into_event()],
-                Some("old-edge".into()),
+        server
+            .sync_room(
+                &client,
+                JoinedRoomBuilder::new(room_id)
+                    .add_timeline_event(f.text_msg("retained").event_id(event_id!("$old")))
+                    .set_timeline_prev_batch("old-edge"),
             )
-            .await
-            .unwrap();
+            .await;
         let gap = cache
             .append_history_section(
                 vec![f.text_msg("recent").event_id(event_id!("$new")).into_event()],
@@ -129,10 +130,13 @@ async fn test_subscribe_with_history_reports_atomic_gap_snapshot_and_updates() {
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
-    cache
-        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
-        .await
-        .unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),
+        )
+        .await;
     cache
         .append_history_section(
             vec![f.reaction(event_id!("$old"), "👍").event_id(event_id!("$hidden")).into_event()],
@@ -195,13 +199,14 @@ async fn test_segmented_history_keeps_unavailable_intervals_and_independent_curs
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
-    cache
-        .reseed(
-            vec![f.text_msg("old").event_id(event_id!("$old")).into_event()],
-            Some("old-edge".into()),
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old")))
+                .set_timeline_prev_batch("old-edge"),
         )
-        .await
-        .unwrap();
+        .await;
     let first = cache
         .append_history_section(
             vec![f.text_msg("middle").event_id(event_id!("$middle")).into_event()],
@@ -261,13 +266,14 @@ async fn test_segmented_history_failed_commit_keeps_events_and_cursor_retryable(
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
-    cache
-        .reseed(
-            vec![f.text_msg("old").event_id(event_id!("$old")).into_event()],
-            Some("old-edge".into()),
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old")))
+                .set_timeline_prev_batch("old-edge"),
         )
-        .await
-        .unwrap();
+        .await;
     store.fail_writes.store(true, Ordering::SeqCst);
     assert!(
         cache
@@ -321,10 +327,13 @@ async fn test_segmented_history_request_failure_preserves_existing_progress() {
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
-    cache
-        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
-        .await
-        .unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),
+        )
+        .await;
     let gap = cache
         .append_history_section(
             vec![f.text_msg("new").event_id(event_id!("$new")).into_event()],
@@ -365,10 +374,13 @@ async fn test_segmented_history_cancellation_keeps_the_saved_cursor() {
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
-    cache
-        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
-        .await
-        .unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),
+        )
+        .await;
     let gap = cache
         .append_history_section(
             vec![f.text_msg("new").event_id(event_id!("$new")).into_event()],
@@ -423,10 +435,13 @@ async fn test_segmented_history_preserves_sync_arrivals_during_a_fill() {
     let room = server.sync_joined_room(&client, room_id).await;
     let (cache, _handles) = room.event_cache().await.unwrap();
     let f = EventFactory::new().room(room_id).sender(user_id!("@alice:example.org"));
-    cache
-        .reseed(vec![f.text_msg("old").event_id(event_id!("$old")).into_event()], None)
-        .await
-        .unwrap();
+    server
+        .sync_room(
+            &client,
+            JoinedRoomBuilder::new(room_id)
+                .add_timeline_event(f.text_msg("old").event_id(event_id!("$old"))),
+        )
+        .await;
     let gap = cache
         .append_history_section(
             vec![f.text_msg("new").event_id(event_id!("$new")).into_event()],

Added file: crates/matrix-sdk/tests/support/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 4fae3b516f05

Commit `4fae3b516f05`