atagen

matrix-rust-sdk

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

Commit 72db3bb7db1c

tarball

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

16 files changed+1306-219
event_cache: clear all room-owned history views

Hashed SQLite cache IDs need persistent ownership and clear markers so legacy child timelines cannot reappear after room replacement.
crates/matrix-sdk-base/src/event_cache/store/integration_tests.rs+186
@@ -197,6 +197,12 @@ pub trait EventCacheStoreIntegrationTests {
197197 /// Test that removing a room from storage empties all associated data.
198198 async fn test_remove_room(&self);
199199
200+ #[allow(missing_docs)]
201+ async fn test_remove_room_clears_all_views(&self);
202+
203+ #[allow(missing_docs)]
204+ async fn test_non_room_clear_keeps_sibling_views(&self);
205+
200206 /// Test that filtering duplicated events works as expected.
201207 async fn test_filter_duplicated_events(&self);
202208
@@ -1379,6 +1385,172 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
13791385 assert!(!r1_linked_chunk.is_empty());
13801386 }
13811387
1388+ async fn test_remove_room_clears_all_views(&self) {
1389+ let room_id = room_id!("!r0:matrix.org");
1390+ let another_room_id = room_id!("!r1:matrix.org");
1391+ let thread_root = event_id!("$thread_root");
1392+ let focused_event_id = event_id!("$focused_event");
1393+
1394+ let room_event_id = event_id!("$room_event");
1395+ let room_event = make_test_event_with_event_id(room_id, "room event", Some(room_event_id));
1396+ let thread_event_id = event_id!("$thread_event");
1397+ let thread_event =
1398+ make_test_event_with_event_id(room_id, "thread event", Some(thread_event_id));
1399+ let pinned_event_id = event_id!("$pinned_event");
1400+ let pinned_event =
1401+ make_test_event_with_event_id(room_id, "pinned event", Some(pinned_event_id));
1402+ let focused_event =
1403+ make_test_event_with_event_id(room_id, "focused event", Some(focused_event_id));
1404+ let standalone_event_id = event_id!("$standalone_event");
1405+ let standalone_event =
1406+ make_test_event_with_event_id(room_id, "standalone event", Some(standalone_event_id));
1407+ let other_room_saved_event_id = event_id!("$other_room_saved_event");
1408+ let other_room_saved_event = make_test_event_with_event_id(
1409+ another_room_id,
1410+ "other room saved event",
1411+ Some(other_room_saved_event_id),
1412+ );
1413+ let other_room_timeline_event_id = event_id!("$other_room_timeline_event");
1414+ let other_room_timeline_event = make_test_event_with_event_id(
1415+ another_room_id,
1416+ "other room timeline event",
1417+ Some(other_room_timeline_event_id),
1418+ );
1419+
1420+ let room_linked_chunk_id = LinkedChunkId::Room(room_id);
1421+ let thread_linked_chunk_id = LinkedChunkId::Thread(room_id, thread_root);
1422+ let pinned_linked_chunk_id = LinkedChunkId::PinnedEvents(room_id);
1423+ let focused_linked_chunk_id = LinkedChunkId::EventFocused(room_id, focused_event_id);
1424+
1425+ for (linked_chunk_id, event) in [
1426+ (room_linked_chunk_id, room_event),
1427+ (thread_linked_chunk_id, thread_event),
1428+ (pinned_linked_chunk_id, pinned_event),
1429+ (focused_linked_chunk_id, focused_event),
1430+ ] {
1431+ self.handle_linked_chunk_updates(
1432+ linked_chunk_id,
1433+ vec![
1434+ Update::NewItemsChunk { previous: None, new: CId::new(0), next: None },
1435+ Update::PushItems { at: Position::new(CId::new(0), 0), items: vec![event] },
1436+ ],
1437+ )
1438+ .await
1439+ .unwrap();
1440+ }
1441+
1442+ self.save_event(room_id, standalone_event).await.unwrap();
1443+ self.save_event(another_room_id, other_room_saved_event).await.unwrap();
1444+ self.handle_linked_chunk_updates(
1445+ LinkedChunkId::Room(another_room_id),
1446+ vec![
1447+ Update::NewItemsChunk { previous: None, new: CId::new(0), next: None },
1448+ Update::PushItems {
1449+ at: Position::new(CId::new(0), 0),
1450+ items: vec![other_room_timeline_event],
1451+ },
1452+ ],
1453+ )
1454+ .await
1455+ .unwrap();
1456+
1457+ self.handle_linked_chunk_updates(room_linked_chunk_id, vec![Update::Clear]).await.unwrap();
1458+
1459+ for linked_chunk_id in [
1460+ room_linked_chunk_id,
1461+ thread_linked_chunk_id,
1462+ pinned_linked_chunk_id,
1463+ focused_linked_chunk_id,
1464+ ] {
1465+ assert!(self.load_all_chunks(linked_chunk_id).await.unwrap().is_empty());
1466+ }
1467+
1468+ for event_id in
1469+ [room_event_id, thread_event_id, pinned_event_id, focused_event_id, standalone_event_id]
1470+ {
1471+ assert!(
1472+ self.find_event(room_id, event_id)
1473+ .await
1474+ .expect("failed to query for finding an event")
1475+ .is_none()
1476+ );
1477+ }
1478+
1479+ let reinserted_event_id = event_id!("$reinserted_event");
1480+ let reinserted_event =
1481+ make_test_event_with_event_id(room_id, "reinserted event", Some(reinserted_event_id));
1482+ self.handle_linked_chunk_updates(
1483+ room_linked_chunk_id,
1484+ vec![
1485+ Update::NewItemsChunk { previous: None, new: CId::new(0), next: None },
1486+ Update::PushItems {
1487+ at: Position::new(CId::new(0), 0),
1488+ items: vec![reinserted_event],
1489+ },
1490+ ],
1491+ )
1492+ .await
1493+ .unwrap();
1494+
1495+ let chunks = self.load_all_chunks(room_linked_chunk_id).await.unwrap();
1496+ assert_eq!(chunks.len(), 1);
1497+ assert!(
1498+ self.find_event(room_id, reinserted_event_id)
1499+ .await
1500+ .expect("failed to query for finding an event")
1501+ .is_some()
1502+ );
1503+
1504+ let another_room_chunks =
1505+ self.load_all_chunks(LinkedChunkId::Room(another_room_id)).await.unwrap();
1506+ assert_eq!(another_room_chunks.len(), 1);
1507+ for event_id in [other_room_saved_event_id, other_room_timeline_event_id] {
1508+ assert!(
1509+ self.find_event(another_room_id, event_id)
1510+ .await
1511+ .expect("failed to query for finding an event")
1512+ .is_some()
1513+ );
1514+ }
1515+ }
1516+
1517+ async fn test_non_room_clear_keeps_sibling_views(&self) {
1518+ let room_id = *DEFAULT_TEST_ROOM_ID;
1519+ let thread_root = event_id!("$thread_root");
1520+ let focused_event_id = event_id!("$focused_event");
1521+ let linked_chunk_ids = [
1522+ LinkedChunkId::Room(room_id),
1523+ LinkedChunkId::Thread(room_id, thread_root),
1524+ LinkedChunkId::PinnedEvents(room_id),
1525+ LinkedChunkId::EventFocused(room_id, focused_event_id),
1526+ ];
1527+
1528+ for linked_chunk_id in linked_chunk_ids {
1529+ self.handle_linked_chunk_updates(
1530+ linked_chunk_id,
1531+ vec![Update::NewItemsChunk { previous: None, new: CId::new(0), next: None }],
1532+ )
1533+ .await
1534+ .unwrap();
1535+ }
1536+
1537+ self.handle_linked_chunk_updates(
1538+ LinkedChunkId::Thread(room_id, thread_root),
1539+ vec![Update::Clear],
1540+ )
1541+ .await
1542+ .unwrap();
1543+
1544+ for linked_chunk_id in linked_chunk_ids {
1545+ let chunks = self.load_all_chunks(linked_chunk_id).await.unwrap();
1546+ if linked_chunk_id == LinkedChunkId::Thread(room_id, thread_root) {
1547+ assert!(chunks.is_empty());
1548+ } else {
1549+ assert_eq!(chunks.len(), 1);
1550+ }
1551+ }
1552+ }
1553+
13821554 async fn test_filter_duplicated_events(&self) {
13831555 let room_id = room_id!("!r0:matrix.org");
13841556 let linked_chunk_id = LinkedChunkId::Room(room_id);
@@ -2437,6 +2609,20 @@ macro_rules! event_cache_store_integration_tests {
24372609 }
24382610
24392611 #[async_test]
2612+ async fn test_remove_room_clears_all_views() {
2613+ let event_cache_store =
2614+ get_event_cache_store().await.unwrap().into_event_cache_store();
2615+ event_cache_store.test_remove_room_clears_all_views().await;
2616+ }
2617+
2618+ #[async_test]
2619+ async fn test_non_room_clear_keeps_sibling_views() {
2620+ let event_cache_store =
2621+ get_event_cache_store().await.unwrap().into_event_cache_store();
2622+ event_cache_store.test_non_room_clear_keeps_sibling_views().await;
2623+ }
2624+
2625+ #[async_test]
24402626 async fn test_filter_duplicated_events() {
24412627 let event_cache_store =
24422628 get_event_cache_store().await.unwrap().into_event_cache_store();
crates/matrix-sdk-common/src/linked_chunk/relational.rs+11−3
@@ -281,9 +281,17 @@ where
281281 Update::StartReattachItems | Update::EndReattachItems => { /* nothing */ }
282282
283283 Update::Clear => {
284- self.chunks.retain(|chunk| chunk.linked_chunk_id != linked_chunk_id);
285- self.items_chunks.retain(|chunk| chunk.linked_chunk_id != linked_chunk_id);
286- // We deliberately leave the items intact.
284+ if let LinkedChunkId::Room(room_id) = linked_chunk_id {
285+ self.chunks.retain(|chunk| chunk.linked_chunk_id.room_id() != room_id);
286+ self.items_chunks
287+ .retain(|chunk| chunk.linked_chunk_id.room_id() != room_id);
288+ self.items
289+ .retain(|linked_chunk_id, _| linked_chunk_id.room_id() != room_id);
290+ } else {
291+ self.chunks.retain(|chunk| chunk.linked_chunk_id != linked_chunk_id);
292+ self.items_chunks.retain(|chunk| chunk.linked_chunk_id != linked_chunk_id);
293+ // We deliberately leave the items intact.
294+ }
287295 }
288296 }
289297 }
crates/matrix-sdk-indexeddb/src/event_cache_store/mod.rs+34−4
@@ -14,7 +14,11 @@
1414
1515 #![cfg_attr(not(test), allow(unused))]
1616
17-use std::{collections::HashMap, rc::Rc, time::Duration};
17+use std::{
18+ collections::{HashMap, HashSet},
19+ rc::Rc,
20+ time::Duration,
21+};
1822
1923 use indexed_db_futures::{Build, database::Database};
2024 #[cfg(target_family = "wasm")]
@@ -263,9 +267,35 @@ impl EventCacheStore for IndexeddbEventCacheStore {
263267 }
264268 Update::Clear => {
265269 trace!(%linked_chunk_id, "clearing room");
266- transaction.delete_chunks_by_linked_chunk_id(linked_chunk_id).await?;
267- transaction.delete_events_by_linked_chunk_id(linked_chunk_id).await?;
268- transaction.delete_gaps_by_linked_chunk_id(linked_chunk_id).await?;
270+ match linked_chunk_id {
271+ LinkedChunkId::Room(room_id) => {
272+ let linked_chunk_ids = transaction
273+ .get_all_items::<types::Chunk>()
274+ .await?
275+ .into_iter()
276+ .filter(|chunk| chunk.linked_chunk_id.room_id() == room_id)
277+ .map(|chunk| chunk.linked_chunk_id)
278+ .collect::<HashSet<_>>();
279+
280+ for linked_chunk_id in linked_chunk_ids {
281+ let linked_chunk_id = linked_chunk_id.as_ref();
282+ transaction
283+ .delete_chunks_by_linked_chunk_id(linked_chunk_id)
284+ .await?;
285+ transaction
286+ .delete_events_by_linked_chunk_id(linked_chunk_id)
287+ .await?;
288+ transaction.delete_gaps_by_linked_chunk_id(linked_chunk_id).await?;
289+ }
290+
291+ transaction.delete_events_in_room(room_id).await?;
292+ }
293+ linked_chunk_id => {
294+ transaction.delete_chunks_by_linked_chunk_id(linked_chunk_id).await?;
295+ transaction.delete_events_by_linked_chunk_id(linked_chunk_id).await?;
296+ transaction.delete_gaps_by_linked_chunk_id(linked_chunk_id).await?;
297+ }
298+ }
269299 }
270300 }
271301 }
crates/matrix-sdk-indexeddb/src/event_cache_store/transaction.rs+5
@@ -502,6 +502,11 @@ impl<'a> IndexeddbEventCacheStoreTransaction<'a> {
502502 self.delete_items_by_linked_chunk_id::<Event, IndexedEventIdKey>(linked_chunk_id).await
503503 }
504504
505+ pub async fn delete_events_in_room(&self, room_id: &RoomId) -> Result<(), TransactionError> {
506+ let range = IndexedKeyRange::all_with_prefix(room_id, self.serializer().inner());
507+ self.delete_items_by_key::<Event, IndexedEventRoomKey>(range).await
508+ }
509+
505510 /// Query IndexedDB for the gap in the given chunk matching the given linked
506511 /// chunk id.
507512 pub async fn get_gap_by_id(
crates/matrix-sdk-indexeddb/src/transaction/mod.rs+17
@@ -120,6 +120,23 @@ impl<'a> Transaction<'a> {
120120 self.transaction.commit().await.map_err(Into::into)
121121 }
122122
123+ pub(crate) async fn get_all_items<T>(&self) -> Result<Vec<T>, TransactionError>
124+ where
125+ T: Indexed,
126+ T::IndexedType: DeserializeOwned,
127+ T::Error: AsyncErrorDeps,
128+ {
129+ let object_store = self.transaction.object_store(T::OBJECT_STORE)?;
130+ let array = object_store.get_all().serde()?.await?;
131+ let mut items = Vec::with_capacity(array.len());
132+ for value in array {
133+ let item = T::from_indexed(value?, self.serializer.inner())
134+ .map_err(|e| TransactionError::Serialization(Box::new(e)))?;
135+ items.push(item);
136+ }
137+ Ok(items)
138+ }
139+
123140 /// Query IndexedDB for items that match the given key range
124141 pub async fn get_items_by_key<T, K>(
125142 &self,
crates/matrix-sdk-sqlite/migrations/event_cache_store/015_linked_chunk_rooms.sql+17
@@ -0,0 +1,17 @@
1+CREATE TABLE "linked_chunk_rooms" (
2+ "linked_chunk_id" BLOB PRIMARY KEY NOT NULL,
3+ "room_id" BLOB NOT NULL
4+)
5+WITHOUT ROWID;
6+
7+CREATE TABLE "cleared_rooms" (
8+ "room_id" BLOB PRIMARY KEY NOT NULL
9+)
10+WITHOUT ROWID;
11+
12+CREATE INDEX "linked_chunk_rooms_room_id" ON "linked_chunk_rooms" ("room_id");
13+
14+INSERT OR IGNORE INTO "linked_chunk_rooms" ("linked_chunk_id", "room_id")
15+SELECT DISTINCT ec."linked_chunk_id", e."room_id"
16+FROM "event_chunks" ec
17+JOIN "events" e ON e."event_id" = ec."event_id";
crates/matrix-sdk-sqlite/src/event_cache_store.rs+420−106
@@ -578,6 +578,17 @@ async fn run_migrations(conn: &SqliteAsyncConn, version: u8) -> Result<()> {
578578 .await?;
579579 }
580580
581+ if version < 15 {
582+ debug!("Upgrading database to version 15");
583+ conn.with_transaction(|txn| {
584+ txn.execute_batch(include_str!(
585+ "../migrations/event_cache_store/015_linked_chunk_rooms.sql"
586+ ))?;
587+ txn.set_db_version(15)
588+ })
589+ .await?;
590+ }
591+
581592 Ok(())
582593 }
583594
@@ -644,9 +655,17 @@ impl EventCacheStore for SqliteEventCacheStore {
644655 let hashed_linked_chunk_id =
645656 self.encode_key(keys::LINKED_CHUNKS, linked_chunk_id.storage_key());
646657 let linked_chunk_id = linked_chunk_id.to_owned();
658+ let hashed_room_id = self.encode_key(keys::LINKED_CHUNKS, linked_chunk_id.room_id());
659+ let is_room_linked_chunk = matches!(linked_chunk_id.as_ref(), LinkedChunkId::Room(_));
647660 let this = self.clone();
648661
649662 with_immediate_transaction(self, move |txn| {
663+ register_linked_chunk_owner(
664+ txn,
665+ &hashed_linked_chunk_id,
666+ &hashed_room_id,
667+ )?;
668+
650669 for up in updates {
651670 match up {
652671 Update::NewItemsChunk { previous, new, next } => {
@@ -958,11 +977,30 @@ impl EventCacheStore for SqliteEventCacheStore {
958977 Update::Clear => {
959978 trace!(%linked_chunk_id, "clearing items");
960979
961- // Remove chunks, and let cascading do its job.
962- txn.execute(
963- "DELETE FROM linked_chunks WHERE linked_chunk_id = ?",
964- (&hashed_linked_chunk_id,),
965- )?;
980+ if is_room_linked_chunk {
981+ txn.execute(
982+ r#"
983+ DELETE FROM linked_chunks
984+ WHERE linked_chunk_id = ?
985+ OR linked_chunk_id IN (
986+ SELECT linked_chunk_id
987+ FROM linked_chunk_rooms
988+ WHERE room_id = ?
989+ )
990+ "#,
991+ (&hashed_linked_chunk_id, &hashed_room_id),
992+ )?;
993+ txn.execute("DELETE FROM events WHERE room_id = ?", [&hashed_room_id])?;
994+ txn.execute(
995+ "INSERT OR IGNORE INTO cleared_rooms(room_id) VALUES (?)",
996+ [&hashed_room_id],
997+ )?;
998+ } else {
999+ txn.execute(
1000+ "DELETE FROM linked_chunks WHERE linked_chunk_id = ?",
1001+ [&hashed_linked_chunk_id],
1002+ )?;
1003+ }
9661004 }
9671005
9681006 Update::StartReattachItems | Update::EndReattachItems => {
@@ -987,13 +1025,18 @@ impl EventCacheStore for SqliteEventCacheStore {
9871025
9881026 let hashed_linked_chunk_id =
9891027 self.encode_key(keys::LINKED_CHUNKS, linked_chunk_id.storage_key());
1028+ let linked_chunk_id = linked_chunk_id.to_owned();
1029+ let hashed_room_id = self.encode_key(keys::LINKED_CHUNKS, linked_chunk_id.room_id());
9901030
9911031 let this = self.clone();
9921032
993- let result = self
994- .read()
995- .await?
996- .with_transaction(move |txn| -> Result<_> {
1033+ let result = with_immediate_transaction(self, move |txn| -> Result<_> {
1034+ register_linked_chunk_owner(
1035+ txn,
1036+ &hashed_linked_chunk_id,
1037+ &hashed_room_id,
1038+ )?;
1039+
9971040 let mut items = Vec::new();
9981041
9991042 // Use `ORDER BY id` to get a deterministic ordering for testing purposes.
@@ -1031,52 +1074,54 @@ impl EventCacheStore for SqliteEventCacheStore {
10311074
10321075 let hashed_linked_chunk_id =
10331076 self.encode_key(keys::LINKED_CHUNKS, linked_chunk_id.storage_key());
1034-
1035- self.read()
1036- .await?
1037- .with_transaction(move |txn| -> Result<_> {
1038- // We want to collect the metadata about each chunk (id, next, previous), and
1039- // for event chunks, the number of events in it. For gaps, the
1040- // number of events is 0, by convention.
1041- //
1042- // We've tried different strategies over time:
1043- // - use a `LEFT JOIN` + `COUNT`, which was extremely inefficient because it
1044- // caused a full table traversal for each chunk, including for gaps which
1045- // don't have any events. This happened in
1046- // https://github.com/matrix-org/matrix-rust-sdk/pull/5225.
1047- // - use a `CASE` statement on the chunk's type: if it's an event chunk, run an
1048- // additional `SELECT` query. It was an immense improvement, but still caused
1049- // one select query per event chunk. This happened in
1050- // https://github.com/matrix-org/matrix-rust-sdk/pull/5411.
1051- //
1052- // The current solution is to run two queries:
1053- // - one to get each chunk and its number of events, by doing a single `SELECT`
1054- // query over the `event_chunks` table, grouping by chunk ids. This gives us a
1055- // list of `(chunk_id, num_events)` pairs, which can be transformed into a
1056- // hashmap.
1057- // - one to get each chunk's metadata (id, previous, next, type) from the
1058- // database with a `SELECT`, and then use the hashmap to get the number of
1059- // events.
1060- //
1061- // This strategy minimizes the number of queries to the database, and keeps them
1062- // super simple, while doing a bit more processing here, which is much faster.
1063-
1064- let num_events_by_chunk_ids = txn
1065- .prepare(
1066- r#"
1077+ let linked_chunk_id = linked_chunk_id.to_owned();
1078+ let hashed_room_id = self.encode_key(keys::LINKED_CHUNKS, linked_chunk_id.room_id());
1079+
1080+ with_immediate_transaction(self, move |txn| -> Result<_> {
1081+ register_linked_chunk_owner(txn, &hashed_linked_chunk_id, &hashed_room_id)?;
1082+
1083+ // We want to collect the metadata about each chunk (id, next, previous), and
1084+ // for event chunks, the number of events in it. For gaps, the
1085+ // number of events is 0, by convention.
1086+ //
1087+ // We've tried different strategies over time:
1088+ // - use a `LEFT JOIN` + `COUNT`, which was extremely inefficient because it
1089+ // caused a full table traversal for each chunk, including for gaps which
1090+ // don't have any events. This happened in
1091+ // https://github.com/matrix-org/matrix-rust-sdk/pull/5225.
1092+ // - use a `CASE` statement on the chunk's type: if it's an event chunk, run an
1093+ // additional `SELECT` query. It was an immense improvement, but still caused
1094+ // one select query per event chunk. This happened in
1095+ // https://github.com/matrix-org/matrix-rust-sdk/pull/5411.
1096+ //
1097+ // The current solution is to run two queries:
1098+ // - one to get each chunk and its number of events, by doing a single `SELECT`
1099+ // query over the `event_chunks` table, grouping by chunk ids. This gives us a
1100+ // list of `(chunk_id, num_events)` pairs, which can be transformed into a
1101+ // hashmap.
1102+ // - one to get each chunk's metadata (id, previous, next, type) from the
1103+ // database with a `SELECT`, and then use the hashmap to get the number of
1104+ // events.
1105+ //
1106+ // This strategy minimizes the number of queries to the database, and keeps them
1107+ // super simple, while doing a bit more processing here, which is much faster.
1108+
1109+ let num_events_by_chunk_ids = txn
1110+ .prepare(
1111+ r#"
10671112 SELECT ec.chunk_id, COUNT(ec.event_id)
10681113 FROM event_chunks as ec
10691114 WHERE ec.linked_chunk_id = ?
10701115 GROUP BY ec.chunk_id
10711116 "#,
1072- )?
1073- .query_map((&hashed_linked_chunk_id,), |row| {
1074- Ok((row.get::<_, u64>(0)?, row.get::<_, usize>(1)?))
1075- })?
1076- .collect::<Result<HashMap<_, _>, _>>()?;
1117+ )?
1118+ .query_map((&hashed_linked_chunk_id,), |row| {
1119+ Ok((row.get::<_, u64>(0)?, row.get::<_, usize>(1)?))
1120+ })?
1121+ .collect::<Result<HashMap<_, _>, _>>()?;
10771122
1078- txn.prepare(
1079- r#"
1123+ txn.prepare(
1124+ r#"
10801125 SELECT
10811126 lc.id,
10821127 lc.previous,
@@ -1085,39 +1130,39 @@ impl EventCacheStore for SqliteEventCacheStore {
10851130 FROM linked_chunks as lc
10861131 WHERE lc.linked_chunk_id = ?
10871132 ORDER BY lc.id"#,
1088- )?
1089- .query_map((&hashed_linked_chunk_id,), |row| {
1090- Ok((
1091- row.get::<_, u64>(0)?,
1092- row.get::<_, Option<u64>>(1)?,
1093- row.get::<_, Option<u64>>(2)?,
1094- row.get::<_, String>(3)?,
1095- ))
1096- })?
1097- .map(|data| -> Result<_> {
1098- let (id, previous, next, chunk_type) = data?;
1133+ )?
1134+ .query_map((&hashed_linked_chunk_id,), |row| {
1135+ Ok((
1136+ row.get::<_, u64>(0)?,
1137+ row.get::<_, Option<u64>>(1)?,
1138+ row.get::<_, Option<u64>>(2)?,
1139+ row.get::<_, String>(3)?,
1140+ ))
1141+ })?
1142+ .map(|data| -> Result<_> {
1143+ let (id, previous, next, chunk_type) = data?;
1144+
1145+ // Note: since a gap has 0 events, an alternative could be to *not* retrieve
1146+ // the chunk type, and just let the hashmap lookup fail for gaps. However,
1147+ // benchmarking shows that this is slightly slower than matching the chunk
1148+ // type (around 1%, so in the realm of noise), so we keep the explicit
1149+ // check instead.
1150+ let num_items = if chunk_type == CHUNK_TYPE_GAP_TYPE_STRING {
1151+ 0
1152+ } else {
1153+ num_events_by_chunk_ids.get(&id).copied().unwrap_or(0)
1154+ };
10991155
1100- // Note: since a gap has 0 events, an alternative could be to *not* retrieve
1101- // the chunk type, and just let the hashmap lookup fail for gaps. However,
1102- // benchmarking shows that this is slightly slower than matching the chunk
1103- // type (around 1%, so in the realm of noise), so we keep the explicit
1104- // check instead.
1105- let num_items = if chunk_type == CHUNK_TYPE_GAP_TYPE_STRING {
1106- 0
1107- } else {
1108- num_events_by_chunk_ids.get(&id).copied().unwrap_or(0)
1109- };
1110-
1111- Ok(ChunkMetadata {
1112- identifier: ChunkIdentifier::new(id),
1113- previous: previous.map(ChunkIdentifier::new),
1114- next: next.map(ChunkIdentifier::new),
1115- num_items,
1116- })
1156+ Ok(ChunkMetadata {
1157+ identifier: ChunkIdentifier::new(id),
1158+ previous: previous.map(ChunkIdentifier::new),
1159+ next: next.map(ChunkIdentifier::new),
1160+ num_items,
11171161 })
1118- .collect::<Result<Vec<_>, _>>()
11191162 })
1120- .await
1163+ .collect::<Result<Vec<_>, _>>()
1164+ })
1165+ .await
11211166 }
11221167
11231168 #[instrument(skip(self))]
@@ -1129,13 +1174,18 @@ impl EventCacheStore for SqliteEventCacheStore {
11291174
11301175 let hashed_linked_chunk_id =
11311176 self.encode_key(keys::LINKED_CHUNKS, linked_chunk_id.storage_key());
1177+ let linked_chunk_id = linked_chunk_id.to_owned();
1178+ let hashed_room_id = self.encode_key(keys::LINKED_CHUNKS, linked_chunk_id.room_id());
11321179
11331180 let this = self.clone();
11341181
1135- self
1136- .read()
1137- .await?
1138- .with_transaction(move |txn| -> Result<_> {
1182+ with_immediate_transaction(self, move |txn| -> Result<_> {
1183+ register_linked_chunk_owner(
1184+ txn,
1185+ &hashed_linked_chunk_id,
1186+ &hashed_room_id,
1187+ )?;
1188+
11391189 // Find the latest chunk identifier to generate a `ChunkIdentifierGenerator`, and count the number of chunks.
11401190 let (observed_max_identifier, number_of_chunks) = txn
11411191 .prepare(
@@ -1225,13 +1275,18 @@ impl EventCacheStore for SqliteEventCacheStore {
12251275
12261276 let hashed_linked_chunk_id =
12271277 self.encode_key(keys::LINKED_CHUNKS, linked_chunk_id.storage_key());
1278+ let linked_chunk_id = linked_chunk_id.to_owned();
1279+ let hashed_room_id = self.encode_key(keys::LINKED_CHUNKS, linked_chunk_id.room_id());
12281280
12291281 let this = self.clone();
12301282
1231- self
1232- .read()
1233- .await?
1234- .with_transaction(move |txn| -> Result<_> {
1283+ with_immediate_transaction(self, move |txn| -> Result<_> {
1284+ register_linked_chunk_owner(
1285+ txn,
1286+ &hashed_linked_chunk_id,
1287+ &hashed_room_id,
1288+ )?;
1289+
12351290 // Find the chunk before the chunk identified by `before_chunk_identifier`.
12361291 let Some((chunk_identifier, previous_chunk, next_chunk, chunk_type)) = txn
12371292 .prepare(
@@ -1305,10 +1360,15 @@ impl EventCacheStore for SqliteEventCacheStore {
13051360 let hashed_linked_chunk_id =
13061361 self.encode_key(keys::LINKED_CHUNKS, linked_chunk_id.storage_key());
13071362 let linked_chunk_id = linked_chunk_id.to_owned();
1363+ let hashed_room_id = self.encode_key(keys::LINKED_CHUNKS, linked_chunk_id.room_id());
1364+
1365+ with_immediate_transaction(self, move |txn| -> Result<_> {
1366+ register_linked_chunk_owner(
1367+ txn,
1368+ &hashed_linked_chunk_id,
1369+ &hashed_room_id,
1370+ )?;
13081371
1309- self.read()
1310- .await?
1311- .with_transaction(move |txn| -> Result<_> {
13121372 txn.chunk_large_query_over(events, None, move |txn, events| {
13131373 let query = format!(
13141374 r#"
@@ -1418,19 +1478,19 @@ impl EventCacheStore for SqliteEventCacheStore {
14181478 let filters = filters.map(ToOwned::to_owned);
14191479 let store = self.clone();
14201480
1421- self.read()
1422- .await?
1423- .with_transaction(move |txn| -> Result<_> {
1424- find_event_relations_transaction(
1425- store,
1426- hashed_room_id,
1427- hashed_linked_chunk_id,
1428- event_id,
1429- filters,
1430- txn,
1431- )
1432- })
1433- .await
1481+ with_immediate_transaction(self, move |txn| -> Result<_> {
1482+ register_linked_chunk_owner(txn, &hashed_linked_chunk_id, &hashed_room_id)?;
1483+
1484+ find_event_relations_transaction(
1485+ store,
1486+ hashed_room_id,
1487+ hashed_linked_chunk_id,
1488+ event_id,
1489+ filters,
1490+ txn,
1491+ )
1492+ })
1493+ .await
14341494 }
14351495
14361496 #[instrument(skip(self))]
@@ -1724,6 +1784,39 @@ fn insert_chunk(
17241784 Ok(())
17251785 }
17261786
1787+fn register_linked_chunk_owner(
1788+ txn: &Transaction<'_>,
1789+ linked_chunk_id: &Key,
1790+ room_id: &Key,
1791+) -> rusqlite::Result<()> {
1792+ let already_registered = txn.query_row(
1793+ "SELECT EXISTS(SELECT 1 FROM linked_chunk_rooms WHERE linked_chunk_id = ?)",
1794+ [linked_chunk_id],
1795+ |row| row.get::<_, bool>(0),
1796+ )?;
1797+
1798+ if already_registered {
1799+ return Ok(());
1800+ }
1801+
1802+ let room_was_cleared = txn.query_row(
1803+ "SELECT EXISTS(SELECT 1 FROM cleared_rooms WHERE room_id = ?)",
1804+ [room_id],
1805+ |row| row.get::<_, bool>(0),
1806+ )?;
1807+
1808+ if room_was_cleared {
1809+ txn.execute("DELETE FROM linked_chunks WHERE linked_chunk_id = ?", [linked_chunk_id])?;
1810+ }
1811+
1812+ txn.execute(
1813+ "INSERT OR IGNORE INTO linked_chunk_rooms(linked_chunk_id, room_id) VALUES (?, ?)",
1814+ (linked_chunk_id, room_id),
1815+ )?;
1816+
1817+ Ok(())
1818+}
1819+
17271820 #[cfg(test)]
17281821 mod tests {
17291822 use std::{
@@ -1736,21 +1829,23 @@ mod tests {
17361829
17371830 use assert_matches::assert_matches;
17381831 use matrix_sdk_base::{
1832+ event_cache::Gap,
17391833 event_cache::store::{
17401834 EventCacheStore, EventCacheStoreError, IntoEventCacheStore,
17411835 integration_tests::EventCacheStoreIntegrationTests,
17421836 },
17431837 event_cache_store_integration_tests, event_cache_store_integration_tests_time,
1744- linked_chunk::{ChunkIdentifier, LinkedChunkId, Update},
1838+ linked_chunk::{ChunkIdentifier, LinkedChunkId, Position, Update},
17451839 };
1746- use matrix_sdk_test::{DEFAULT_TEST_ROOM_ID, async_test};
1840+ use matrix_sdk_test::{DEFAULT_TEST_ROOM_ID, async_test, event_factory::EventFactory};
1841+ use ruma::{event_id, room_id, user_id};
17471842 use tempfile::{TempDir, tempdir};
17481843
17491844 use super::SqliteEventCacheStore;
17501845 use crate::{
17511846 SqliteStoreConfig,
17521847 event_cache_store::keys,
1753- utils::{EncryptableStore as _, SqliteAsyncConnExt},
1848+ utils::{EncryptableStore as _, SqliteAsyncConnExt, SqliteKeyValueStoreConnExt},
17541849 };
17551850
17561851 static TMP_DIR: LazyLock<TempDir> = LazyLock::new(|| tempdir().unwrap());
@@ -1869,6 +1964,225 @@ mod tests {
18691964 }
18701965
18711966 #[async_test]
1967+ async fn test_room_clear_removes_room_views_and_events_only() {
1968+ let store = get_event_cache_store().await.unwrap();
1969+ let room_id = room_id!("!clear:localhost");
1970+ let other_room_id = room_id!("!keep:localhost");
1971+ let room_linked_chunk_id = LinkedChunkId::Room(room_id);
1972+ let auxiliary_linked_chunk_id =
1973+ LinkedChunkId::Thread(room_id, event_id!("$thread:localhost"));
1974+ let other_linked_chunk_id = LinkedChunkId::PinnedEvents(other_room_id);
1975+
1976+ for (linked_chunk_id, chunk_id) in [(room_linked_chunk_id, 1), (other_linked_chunk_id, 3)] {
1977+ store
1978+ .handle_linked_chunk_updates(
1979+ linked_chunk_id,
1980+ vec![Update::NewGapChunk {
1981+ previous: None,
1982+ new: ChunkIdentifier::new(chunk_id),
1983+ next: None,
1984+ gap: Gap { token: "gap".to_owned() },
1985+ }],
1986+ )
1987+ .await
1988+ .unwrap();
1989+ }
1990+
1991+ let sender = user_id!("@alice:localhost");
1992+ let auxiliary_event = EventFactory::new()
1993+ .room(room_id)
1994+ .sender(sender)
1995+ .text_msg("auxiliary event")
1996+ .event_id(event_id!("$aux-event:localhost"))
1997+ .into_event();
1998+ store
1999+ .handle_linked_chunk_updates(
2000+ auxiliary_linked_chunk_id,
2001+ vec![
2002+ Update::NewItemsChunk {
2003+ previous: None,
2004+ new: ChunkIdentifier::new(2),
2005+ next: None,
2006+ },
2007+ Update::PushItems {
2008+ at: Position::new(ChunkIdentifier::new(2), 0),
2009+ items: vec![auxiliary_event],
2010+ },
2011+ ],
2012+ )
2013+ .await
2014+ .unwrap();
2015+
2016+ let room_event_id = event_id!("$room-event:localhost");
2017+ let other_event_id = event_id!("$other-event:localhost");
2018+ let room_event = EventFactory::new()
2019+ .room(room_id)
2020+ .sender(sender)
2021+ .text_msg("room event")
2022+ .event_id(room_event_id)
2023+ .into_event();
2024+ let other_event = EventFactory::new()
2025+ .room(other_room_id)
2026+ .sender(sender)
2027+ .text_msg("other event")
2028+ .event_id(other_event_id)
2029+ .into_event();
2030+ store.save_event(room_id, room_event).await.unwrap();
2031+ store.save_event(other_room_id, other_event).await.unwrap();
2032+
2033+ store.handle_linked_chunk_updates(room_linked_chunk_id, vec![Update::Clear]).await.unwrap();
2034+
2035+ assert!(store.load_all_chunks(room_linked_chunk_id).await.unwrap().is_empty());
2036+ assert!(store.load_all_chunks(auxiliary_linked_chunk_id).await.unwrap().is_empty());
2037+ assert_eq!(store.load_all_chunks(other_linked_chunk_id).await.unwrap().len(), 1);
2038+ assert!(store.find_event(room_id, room_event_id).await.unwrap().is_none());
2039+ assert!(
2040+ store.find_event(room_id, event_id!("$aux-event:localhost")).await.unwrap().is_none()
2041+ );
2042+ assert!(store.find_event(other_room_id, other_event_id).await.unwrap().is_some());
2043+ }
2044+
2045+ #[async_test]
2046+ async fn test_room_clear_drops_legacy_unregistered_gap_view_on_access() {
2047+ let store = get_event_cache_store().await.unwrap();
2048+ let room_id = room_id!("!legacy:localhost");
2049+ let room_linked_chunk_id = LinkedChunkId::Room(room_id);
2050+ let legacy_linked_chunk_id = LinkedChunkId::PinnedEvents(room_id);
2051+ let hashed_legacy_linked_chunk_id =
2052+ store.encode_key(keys::LINKED_CHUNKS, legacy_linked_chunk_id.storage_key());
2053+ let encoded_token = store.encode_value(serde_json::to_vec("old gap").unwrap()).unwrap();
2054+
2055+ store
2056+ .write()
2057+ .await
2058+ .unwrap()
2059+ .with_transaction(move |txn| -> rusqlite::Result<()> {
2060+ txn.execute(
2061+ "INSERT INTO linked_chunks(linked_chunk_id, id, previous, next, type) VALUES (?, ?, NULL, NULL, 'G')",
2062+ (&hashed_legacy_linked_chunk_id, 1),
2063+ )?;
2064+ txn.execute(
2065+ "INSERT INTO gap_chunks(chunk_id, linked_chunk_id, prev_token) VALUES (?, ?, ?)",
2066+ (1, &hashed_legacy_linked_chunk_id, encoded_token),
2067+ )?;
2068+ Ok(())
2069+ })
2070+ .await
2071+ .unwrap();
2072+
2073+ store.handle_linked_chunk_updates(room_linked_chunk_id, vec![Update::Clear]).await.unwrap();
2074+
2075+ assert!(store.load_all_chunks(legacy_linked_chunk_id).await.unwrap().is_empty());
2076+
2077+ store
2078+ .handle_linked_chunk_updates(
2079+ legacy_linked_chunk_id,
2080+ vec![Update::NewGapChunk {
2081+ previous: None,
2082+ new: ChunkIdentifier::new(2),
2083+ next: None,
2084+ gap: Gap { token: "new gap".to_owned() },
2085+ }],
2086+ )
2087+ .await
2088+ .unwrap();
2089+
2090+ assert_eq!(store.load_all_chunks(legacy_linked_chunk_id).await.unwrap().len(), 1);
2091+ }
2092+
2093+ #[async_test]
2094+ async fn test_room_clear_uses_migration_backfilled_owners() {
2095+ let store = get_event_cache_store().await.unwrap();
2096+ let room_id = room_id!("!migration:localhost");
2097+ let other_room_id = room_id!("!legacy-gap:localhost");
2098+ let auxiliary_linked_chunk_id = LinkedChunkId::PinnedEvents(room_id);
2099+ let other_legacy_linked_chunk_id = LinkedChunkId::PinnedEvents(other_room_id);
2100+ let event_id = event_id!("$migration-event:localhost");
2101+ let sender = user_id!("@alice:localhost");
2102+ let event = EventFactory::new()
2103+ .room(room_id)
2104+ .sender(sender)
2105+ .text_msg("migration event")
2106+ .event_id(event_id)
2107+ .into_event();
2108+
2109+ store
2110+ .handle_linked_chunk_updates(
2111+ auxiliary_linked_chunk_id,
2112+ vec![
2113+ Update::NewItemsChunk {
2114+ previous: None,
2115+ new: ChunkIdentifier::new(1),
2116+ next: None,
2117+ },
2118+ Update::PushItems {
2119+ at: Position::new(ChunkIdentifier::new(1), 0),
2120+ items: vec![event],
2121+ },
2122+ ],
2123+ )
2124+ .await
2125+ .unwrap();
2126+
2127+ let hashed_legacy_linked_chunk_id =
2128+ store.encode_key(keys::LINKED_CHUNKS, other_legacy_linked_chunk_id.storage_key());
2129+ let encoded_token = store.encode_value(serde_json::to_vec("legacy gap").unwrap()).unwrap();
2130+ store
2131+ .write()
2132+ .await
2133+ .unwrap()
2134+ .with_transaction(move |txn| -> rusqlite::Result<()> {
2135+ txn.execute(
2136+ "INSERT INTO linked_chunks(linked_chunk_id, id, previous, next, type) VALUES (?, ?, NULL, NULL, 'G')",
2137+ (&hashed_legacy_linked_chunk_id, 2),
2138+ )?;
2139+ txn.execute(
2140+ "INSERT INTO gap_chunks(chunk_id, linked_chunk_id, prev_token) VALUES (?, ?, ?)",
2141+ (2, &hashed_legacy_linked_chunk_id, encoded_token),
2142+ )?;
2143+ Ok(())
2144+ })
2145+ .await
2146+ .unwrap();
2147+
2148+ let hashed_auxiliary_linked_chunk_id =
2149+ store.encode_key(keys::LINKED_CHUNKS, auxiliary_linked_chunk_id.storage_key());
2150+ let hashed_room_id = store.encode_key(keys::LINKED_CHUNKS, room_id);
2151+ {
2152+ let conn = store.write().await.unwrap();
2153+ conn.with_transaction(|txn| -> rusqlite::Result<()> {
2154+ txn.execute_batch("DROP TABLE linked_chunk_rooms; DROP TABLE cleared_rooms;")?;
2155+ txn.set_db_version(14)
2156+ })
2157+ .await
2158+ .unwrap();
2159+
2160+ super::run_migrations(&conn, 14).await.unwrap();
2161+
2162+ let owner_was_backfilled = conn
2163+ .with_transaction(move |txn| {
2164+ txn.query_row(
2165+ "SELECT EXISTS(SELECT 1 FROM linked_chunk_rooms WHERE linked_chunk_id = ? AND room_id = ?)",
2166+ (&hashed_auxiliary_linked_chunk_id, &hashed_room_id),
2167+ |row| row.get::<_, bool>(0),
2168+ )
2169+ })
2170+ .await
2171+ .unwrap();
2172+ assert!(owner_was_backfilled);
2173+ }
2174+
2175+ store
2176+ .handle_linked_chunk_updates(LinkedChunkId::Room(room_id), vec![Update::Clear])
2177+ .await
2178+ .unwrap();
2179+
2180+ assert!(store.load_all_chunks(auxiliary_linked_chunk_id).await.unwrap().is_empty());
2181+ assert!(store.find_event(room_id, event_id).await.unwrap().is_none());
2182+ assert_eq!(store.load_all_chunks(other_legacy_linked_chunk_id).await.unwrap().len(), 1);
2183+ }
2184+
2185+ #[async_test]
18722186 async fn test_linked_chunk_update_is_a_transaction() {
18732187 let store = get_event_cache_store().await.expect("creating cache store failed");
18742188
crates/matrix-sdk/src/event_cache/caches/event_focused/mod.rs+112−17
@@ -29,12 +29,16 @@
2929 //! case where we'd want to persist these caches on disk (e.g., for permalinks
3030 //! to work across sessions).
3131
32-use std::sync::Arc;
32+use std::sync::{
33+ Arc, Mutex,
34+ atomic::{AtomicU64, Ordering},
35+};
3336
37+use eyeball_im::VectorDiff;
3438 use matrix_sdk_base::{
3539 deserialized_responses::TimelineEvent,
3640 event_cache::{Event, Gap},
37- linked_chunk::OwnedLinkedChunkId,
41+ linked_chunk::{OwnedLinkedChunkId, Update},
3842 };
3943 use matrix_sdk_common::{linked_chunk::ChunkIdentifier, serde_helpers::extract_thread_root};
4044 use ruma::{OwnedEventId, UInt, api::Direction};
@@ -91,6 +95,10 @@ pub(crate) enum EventFocusedPaginationMode {
9195 }
9296
9397 struct EventFocusedCacheInner {
98+ generation: Arc<AtomicU64>,
99+ invalidation_lock: Arc<Mutex<()>>,
100+ loaded_generation: u64,
101+
94102 /// The room owning this event-focused cache.
95103 room: WeakRoom,
96104
@@ -111,6 +119,16 @@ struct EventFocusedCacheInner {
111119 }
112120
113121 impl EventFocusedCacheInner {
122+ fn synchronize_generation(&mut self) {
123+ let generation = self.generation.load(Ordering::SeqCst);
124+ if generation != self.loaded_generation {
125+ self.chunk.reset();
126+ let _ = self.chunk.store_updates().take();
127+ let _ = self.chunk.updates_as_vector_diffs();
128+ self.loaded_generation = generation;
129+ }
130+ }
131+
114132 /// Initialize the cache from a focused event.
115133 ///
116134 /// This uses `/context` to fetch the event with surrounding context.
@@ -305,6 +323,13 @@ impl EventFocusedCacheInner {
305323 /// there are more events to fetch.
306324 #[instrument(skip(self), fields(room_id = %self.room.room_id()))]
307325 async fn paginate_backwards(&mut self, num_events: u16) -> Result<PaginationResult> {
326+ let invalidation_lock = self.invalidation_lock.clone();
327+ let generation = {
328+ let _guard = invalidation_lock.lock().unwrap();
329+ self.synchronize_generation();
330+ self.loaded_generation
331+ };
332+
308333 let room = self.room.get().ok_or(EventCacheError::ClientDropped)?;
309334
310335 // Find the gap at the front (backward pagination token).
@@ -318,15 +343,23 @@ impl EventFocusedCacheInner {
318343 trace!(?token, "paginating backwards with token from front gap");
319344
320345 // Fetch events based on pagination mode.
321- let (mut events, new_token) = match &self.pagination_mode {
346+ let pagination = match &self.pagination_mode {
322347 EventFocusedPaginationMode::Room { .. } => {
323- Self::fetch_room_backwards(&room, num_events, &token).await?
348+ Self::fetch_room_backwards(&room, num_events, &token).await
324349 }
325350 EventFocusedPaginationMode::Thread { thread_root } => {
326- Self::fetch_thread_backwards(&room, num_events, &token, thread_root.clone()).await?
351+ Self::fetch_thread_backwards(&room, num_events, &token, thread_root.clone()).await
327352 }
328353 };
329354
355+ let _invalidation_guard = invalidation_lock.lock().unwrap();
356+ if generation != self.generation.load(Ordering::SeqCst) {
357+ self.synchronize_generation();
358+ return Ok(PaginationResult { events: Vec::new(), hit_end_of_timeline: true });
359+ }
360+
361+ let (mut events, new_token) = pagination?;
362+
330363 // Events are in the reverse order, per the API contracts defined in the two
331364 // fetch methods.
332365 events.reverse();
@@ -417,6 +450,13 @@ impl EventFocusedCacheInner {
417450 /// there are more events to fetch.
418451 #[instrument(skip(self), fields(room_id = %self.room.room_id()))]
419452 async fn paginate_forwards(&mut self, num_events: u16) -> Result<PaginationResult> {
453+ let invalidation_lock = self.invalidation_lock.clone();
454+ let generation = {
455+ let _guard = invalidation_lock.lock().unwrap();
456+ self.synchronize_generation();
457+ self.loaded_generation
458+ };
459+
420460 let room = self.room.get().ok_or(EventCacheError::ClientDropped)?;
421461
422462 // Find the gap at the back (forward pagination token).
@@ -430,15 +470,23 @@ impl EventFocusedCacheInner {
430470 trace!(?token, "paginating forwards with token from back gap");
431471
432472 // Fetch events based on pagination mode.
433- let (events, new_token) = match &self.pagination_mode {
473+ let pagination = match &self.pagination_mode {
434474 EventFocusedPaginationMode::Room { .. } => {
435- Self::fetch_room_forwards(&room, num_events, &token).await?
475+ Self::fetch_room_forwards(&room, num_events, &token).await
436476 }
437477 EventFocusedPaginationMode::Thread { thread_root } => {
438- Self::fetch_thread_forwards(&room, num_events, &token, thread_root.clone()).await?
478+ Self::fetch_thread_forwards(&room, num_events, &token, thread_root.clone()).await
439479 }
440480 };
441481
482+ let _invalidation_guard = invalidation_lock.lock().unwrap();
483+ if generation != self.generation.load(Ordering::SeqCst) {
484+ self.synchronize_generation();
485+ return Ok(PaginationResult { events: Vec::new(), hit_end_of_timeline: true });
486+ }
487+
488+ let (events, new_token) = pagination?;
489+
442490 let hit_end = new_token.is_none();
443491 let new_gap = new_token.map(|t| Gap { token: t });
444492
@@ -521,6 +569,11 @@ impl EventFocusedCacheInner {
521569 #[derive(Clone)]
522570 pub struct EventFocusedCache {
523571 inner: Arc<RwLock<EventFocusedCacheInner>>,
572+ generation: Arc<AtomicU64>,
573+ invalidation_lock: Arc<Mutex<()>>,
574+ sender: Sender<TimelineVectorDiffs>,
575+ linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
576+ linked_chunk_id: OwnedLinkedChunkId,
524577 }
525578
526579 impl EventFocusedCache {
@@ -530,36 +583,71 @@ impl EventFocusedCache {
530583 focused_event_id: OwnedEventId,
531584 linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
532585 ) -> Self {
586+ let generation = Arc::new(AtomicU64::new(0));
587+ let invalidation_lock = Arc::new(Mutex::new(()));
588+ let sender = Sender::new(32);
589+ let linked_chunk_id =
590+ OwnedLinkedChunkId::EventFocused(room.room_id().to_owned(), focused_event_id.clone());
591+
533592 Self {
534593 inner: Arc::new(RwLock::new(EventFocusedCacheInner {
594+ generation: generation.clone(),
595+ invalidation_lock: invalidation_lock.clone(),
596+ loaded_generation: 0,
535597 room,
536598 focused_event_id,
537599 pagination_mode: EventFocusedPaginationMode::Room { hide_thread_events: false },
538600 chunk: EventLinkedChunk::new(),
539- sender: Sender::new(32),
540- linked_chunk_update_sender,
601+ sender: sender.clone(),
602+ linked_chunk_update_sender: linked_chunk_update_sender.clone(),
541603 })),
604+ generation,
605+ invalidation_lock,
606+ sender,
607+ linked_chunk_update_sender,
608+ linked_chunk_id,
542609 }
543610 }
544611
612+ pub(super) fn clear(&self) {
613+ let _guard = self.invalidation_lock.lock().unwrap();
614+ self.generation.fetch_add(1, Ordering::SeqCst);
615+ let _ = self.sender.send(TimelineVectorDiffs {
616+ diffs: vec![VectorDiff::Clear],
617+ origin: EventsOrigin::Cache,
618+ });
619+ let _ = self.linked_chunk_update_sender.send(RoomEventCacheLinkedChunkUpdate {
620+ updates: vec![Update::Clear],
621+ linked_chunk_id: self.linked_chunk_id.clone(),
622+ });
623+ }
624+
545625 /// Subscribe to updates from this event-focused timeline.
546626 pub async fn subscribe(&self) -> (Vec<Event>, Receiver<TimelineVectorDiffs>) {
547- let inner = self.inner.read().await;
548- let events = inner.chunk.events().map(|(_position, item)| item.clone()).collect();
627+ let mut inner = self.inner.write().await;
628+ let _guard = self.invalidation_lock.lock().unwrap();
629+ inner.synchronize_generation();
549630 let recv = inner.sender.subscribe();
631+ let events = inner.chunk.events().map(|(_position, item)| item.clone()).collect();
550632 (events, recv)
551633 }
552634
553635 /// Check if we've hit the start of the timeline (no more backward
554636 /// pagination possible).
555637 pub async fn hit_timeline_start(&self) -> bool {
556- self.inner.read().await.first_chunk_as_gap().is_none()
638+ let mut inner = self.inner.write().await;
639+ let _guard = self.invalidation_lock.lock().unwrap();
640+ inner.synchronize_generation();
641+ inner.first_chunk_as_gap().is_none()
557642 }
558643
559644 /// Check if we've hit the end of the timeline (no more forward pagination
560645 /// possible).
561646 pub async fn hit_timeline_end(&self) -> bool {
562- self.inner.read().await.last_chunk_as_gap().is_none()
647+ let mut inner = self.inner.write().await;
648+ let _guard = self.invalidation_lock.lock().unwrap();
649+ inner.synchronize_generation();
650+ inner.last_chunk_as_gap().is_none()
563651 }
564652
565653 /// Start the event-focused timeline from the focused event, fetching
@@ -576,18 +664,23 @@ impl EventFocusedCache {
576664 /// Paginate backwards in this event-focused timeline, be it room or thread
577665 /// pagination depending on the mode.
578666 pub async fn paginate_backwards(&self, num_events: u16) -> Result<PaginationResult> {
579- self.inner.write().await.paginate_backwards(num_events).await
667+ let mut inner = self.inner.write().await;
668+ inner.paginate_backwards(num_events).await
580669 }
581670
582671 /// Paginate forwards in this event-focused timeline, be it room or thread
583672 /// pagination depending on the mode.
584673 pub async fn paginate_forwards(&self, num_events: u16) -> Result<PaginationResult> {
585- self.inner.write().await.paginate_forwards(num_events).await
674+ let mut inner = self.inner.write().await;
675+ inner.paginate_forwards(num_events).await
586676 }
587677
588678 /// Get the thread root event ID if this linked chunk is in thread mode.
589679 pub async fn thread_root(&self) -> Option<OwnedEventId> {
590- match &self.inner.read().await.pagination_mode {
680+ let mut inner = self.inner.write().await;
681+ let _guard = self.invalidation_lock.lock().unwrap();
682+ inner.synchronize_generation();
683+ match &inner.pagination_mode {
591684 EventFocusedPaginationMode::Thread { thread_root } => Some(thread_root.clone()),
592685 _ => None,
593686 }
@@ -599,6 +692,8 @@ impl EventFocusedCache {
599692 #[cfg(feature = "e2e-encryption")]
600693 pub async fn replace_utds(&self, events: &[ResolvedUtd]) {
601694 let mut guard = self.inner.write().await;
695+ let _invalidation_guard = self.invalidation_lock.lock().unwrap();
696+ guard.synchronize_generation();
602697 if guard.chunk.replace_utds(events) {
603698 guard.propagate_changes();
604699 guard.notify_subscribers(EventsOrigin::Cache);
crates/matrix-sdk/src/event_cache/caches/mod.rs+5
@@ -12,6 +12,8 @@
1212 // See the License for the specific language governing permissions and
1313 // limitations under the License.
1414
15+use std::sync::{Arc, atomic::AtomicU64};
16+
1517 use eyeball::SharedObservable;
1618 use eyeball_im::VectorDiff;
1719 use matrix_sdk_base::{
@@ -68,6 +70,7 @@ impl Caches {
6870 let pagination_status = SharedObservable::new(pagination::SharedPaginationStatus::Idle {
6971 hit_timeline_start: false,
7072 });
73+ let generation = Arc::new(AtomicU64::new(0));
7174
7275 let enabled_thread_support =
7376 matches!(client.base_client().threading_support, ThreadingSupport::Enabled { .. });
@@ -87,6 +90,7 @@ impl Caches {
8790 linked_chunk_update_sender,
8891 store,
8992 pagination_status.clone(),
93+ generation.clone(),
9094 automatic_pagination,
9195 )
9296 .await?;
@@ -99,6 +103,7 @@ impl Caches {
99103 weak_room,
100104 room_state,
101105 pagination_status,
106+ generation,
102107 auto_shrink_sender,
103108 update_sender,
104109 );
crates/matrix-sdk/src/event_cache/caches/pinned_events/mod.rs+33−2
@@ -60,6 +60,8 @@ pub(in super::super) struct PinnedEventCacheState {
6060 /// relations over time.
6161 chunk: EventLinkedChunk,
6262
63+ generation: u64,
64+
6365 /// Reference to the underlying backing store.
6466 // TODO: can be removed?
6567 store: EventCacheStoreLock,
@@ -224,8 +226,14 @@ impl PinnedEventCache {
224226
225227 let chunk = EventLinkedChunk::new();
226228
227- let state =
228- PinnedEventCacheState { room_id, chunk, sender, linked_chunk_update_sender, store };
229+ let state = PinnedEventCacheState {
230+ room_id,
231+ chunk,
232+ generation: 0,
233+ sender,
234+ linked_chunk_update_sender,
235+ store,
236+ };
229237 let state = Arc::new(PinnedEventCacheStateLock::new_inner(state));
230238
231239 let task = Arc::new(
@@ -251,6 +259,17 @@ impl PinnedEventCache {
251259 Ok((events, recv))
252260 }
253261
262+ pub(super) async fn clear(&self) -> Result<()> {
263+ self._task.abort();
264+ let mut guard = self.state.write().await?;
265+ guard.state.generation += 1;
266+ guard.state.chunk.reset();
267+ guard.propagate_changes().await?;
268+ guard.notify_subscribers(EventsOrigin::Cache);
269+
270+ Ok(())
271+ }
272+
254273 /// Try to locate the events in the linked chunk corresponding to the given
255274 /// list of decrypted events, and replace them, while alerting observers
256275 /// about the update.
@@ -362,6 +381,14 @@ impl PinnedEventCache {
362381 debug!("pinned events listener task started");
363382
364383 let reload_from_network = async |room: Room| {
384+ let generation = match state.read().await {
385+ Ok(guard) => guard.state.generation,
386+ Err(err) => {
387+ warn!("error when acquiring read lock to load pinned events: {err}");
388+ return;
389+ }
390+ };
391+
365392 let events = match Self::reload_pinned_events(room).await {
366393 Ok(Some(events)) => events,
367394 Ok(None) => Vec::new(),
@@ -375,6 +402,10 @@ impl PinnedEventCache {
375402 // to the observers.
376403 match state.write().await {
377404 Ok(mut guard) => {
405+ if guard.state.generation != generation {
406+ return;
407+ }
408+
378409 guard.replace_all_events(events).await.unwrap_or_else(|err| {
379410 warn!("error when replacing pinned events: {err}");
380411 });
crates/matrix-sdk/src/event_cache/caches/room/mod.rs+57−55
@@ -26,7 +26,7 @@ use std::{
2626 },
2727 };
2828
29-use eyeball::{ObservableWriteGuard, SharedObservable};
29+use eyeball::SharedObservable;
3030 use matrix_sdk_base::{
3131 deserialized_responses::AmbiguityChange,
3232 event_cache::Event,
@@ -82,6 +82,7 @@ impl RoomEventCache {
8282 weak_room: WeakRoom,
8383 state: LockedRoomEventCacheState,
8484 shared_pagination_status: SharedObservable<SharedPaginationStatus>,
85+ generation: Arc<AtomicU64>,
8586 auto_shrink_sender: mpsc::Sender<AutoShrinkChannelPayload>,
8687 update_sender: RoomEventCacheUpdateSender,
8788 ) -> Self {
@@ -91,6 +92,7 @@ impl RoomEventCache {
9192 weak_room,
9293 state,
9394 shared_pagination_status,
95+ generation,
9496 auto_shrink_sender,
9597 update_sender,
9698 )),
@@ -186,45 +188,53 @@ impl RoomEventCache {
186188 num_context_events: u16,
187189 thread_mode: EventFocusThreadMode,
188190 ) -> Result<EventFocusedCache> {
189- let room = self.inner.weak_room.get().ok_or(EventCacheError::ClientDropped)?;
190- let guard = self.inner.state.read().await?;
191+ loop {
192+ let room = self.inner.weak_room.get().ok_or(EventCacheError::ClientDropped)?;
193+ let guard = self.inner.state.read().await?;
194+
195+ // Check if we already have a cache for this event.
196+ if let Some(cache) = guard.get_event_focused_cache(event_id.clone(), thread_mode) {
197+ trace!("the cache was already created, returning it");
198+ return Ok(cache);
199+ }
191200
192- // Check if we already have a cache for this event.
193- if let Some(cache) = guard.get_event_focused_cache(event_id.clone(), thread_mode) {
194- trace!("the cache was already created, returning it");
195- return Ok(cache);
196- }
201+ // Create a new cache.
202+ let linked_chunk_update_sender = guard.state.linked_chunk_update_sender.clone();
203+ let generation = self.inner.generation.load(Ordering::SeqCst);
197204
198- // Create a new cache.
199- let linked_chunk_update_sender = guard.state.linked_chunk_update_sender.clone();
205+ // Make sure to drop the guard before calling `start_from` below, as it may need
206+ // to lock the room event cache's state again, when memoizing events
207+ // received from the network response.
208+ drop(guard);
200209
201- // Make sure to drop the guard before calling `start_from` below, as it may need
202- // to lock the room event cache's state again, when memoizing events
203- // received from the network response.
204- drop(guard);
210+ let room_id = room.room_id().to_owned();
211+ let weak_room = WeakRoom::new(WeakClient::from_client(&room.client()), room_id.clone());
205212
206- let room_id = room.room_id().to_owned();
207- let weak_room = WeakRoom::new(WeakClient::from_client(&room.client()), room_id.clone());
213+ trace!("creating a fresh event-focused cache");
214+ let cache =
215+ EventFocusedCache::new(weak_room, event_id.clone(), linked_chunk_update_sender);
208216
209- trace!("creating a fresh event-focused cache");
210- let cache = EventFocusedCache::new(weak_room, event_id.clone(), linked_chunk_update_sender);
217+ // Initialize the cache from the server.
218+ cache.start_from(room, num_context_events, thread_mode).await?;
211219
212- // Initialize the cache from the server.
213- cache.start_from(room, num_context_events, thread_mode).await?;
220+ let mut guard = self.inner.state.write().await?;
221+ if self.inner.generation.load(Ordering::SeqCst) != generation {
222+ cache.clear();
223+ continue;
224+ }
214225
215- let mut guard = self.inner.state.write().await?;
226+ // Check again if we already have a cache for this event, just in case there was
227+ // a race with another caller during initialization.
228+ if let Some(cache) = guard.get_event_focused_cache(event_id.clone(), thread_mode) {
229+ trace!("another cache has been racily created, returning it");
230+ return Ok(cache);
231+ }
232+
233+ // Insert the cache in the map.
234+ guard.insert_event_focused_cache(event_id, thread_mode, cache.clone());
216235
217- // Check again if we already have a cache for this event, just in case there was
218- // a race with another caller during initialization.
219- if let Some(cache) = guard.get_event_focused_cache(event_id.clone(), thread_mode) {
220- trace!("another cache has been racily created, returning it");
221236 return Ok(cache);
222237 }
223-
224- // Insert the cache in the map.
225- guard.insert_event_focused_cache(event_id, thread_mode, cache.clone());
226-
227- Ok(cache)
228238 }
229239
230240 /// Get an event-focused cache for this event and thread mode, if it exists.
@@ -333,22 +343,21 @@ impl RoomEventCache {
333343 self.inner.state.read().await?.find_event_relations(event_id, filter.clone()).await
334344 }
335345
336- /// Clear all the storage for this [`RoomEventCache`].
346+ /// Clear this room's cached events and all its cached timeline views.
337347 ///
338- /// This will get rid of all the events from the linked chunk and persisted
339- /// storage.
348+ /// This removes persisted room, thread, pinned, and event-focused history.
349+ /// Account credentials, encryption state, and pending sends are preserved.
340350 pub async fn clear(&self) -> Result<()> {
351+ let cache = self.clone();
352+ matrix_sdk_common::executor::spawn(async move { cache.clear_inner().await })
353+ .await
354+ .expect("joining room cache clear failed")
355+ }
356+
357+ async fn clear_inner(&self) -> Result<()> {
341358 // Clear the linked chunk and persisted storage.
342359 let mut state = self.inner.state.write().await?;
343360 let updates_as_vector_diffs = state.reset().await?;
344- {
345- let mut status = self.inner.shared_pagination_status.write();
346- self.inner.generation.fetch_add(1, Ordering::SeqCst);
347- ObservableWriteGuard::set(
348- &mut status,
349- SharedPaginationStatus::Idle { hit_timeline_start: false },
350- );
351- }
352361
353362 // Notify observers about the update.
354363 self.inner.update_sender.send(
@@ -368,18 +377,10 @@ impl RoomEventCache {
368377 matrix_sdk_common::executor::spawn(async move {
369378 let mut state = cache.inner.state.write().await?;
370379 state.reseed(events, prev_batch.clone()).await?;
371- {
372- let mut status = cache.inner.shared_pagination_status.write();
373- cache.inner.generation.fetch_add(1, Ordering::SeqCst);
374- ObservableWriteGuard::set(
375- &mut status,
376- SharedPaginationStatus::Idle { hit_timeline_start: prev_batch.is_none() },
377- );
378- }
379380 let events =
380381 state.room_linked_chunk().events().map(|(_, event)| event.clone()).collect();
381382 let result = async {
382- state.clear_loaded_threads().await?;
383+ state.clear_loaded_views().await?;
383384 state
384385 .post_process_new_events(events, PostProcessingOrigin::Backpagination, None)
385386 .await
@@ -387,6 +388,7 @@ impl RoomEventCache {
387388 .await;
388389 let items =
389390 state.room_linked_chunk().events().map(|(_, event)| event.clone()).collect();
391+ let _ = state.room_linked_chunk_mut().updates_as_vector_diffs();
390392 cache.inner.update_sender.send(
391393 RoomEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
392394 diffs: vec![eyeball_im::VectorDiff::Reset { values: items }],
@@ -476,7 +478,7 @@ impl RoomEventCache {
476478
477479 /// The (non-cloneable) details of the `RoomEventCache`.
478480 pub(super) struct RoomEventCacheInner {
479- pub generation: AtomicU64,
481+ pub generation: Arc<AtomicU64>,
480482 /// The room id for this room.
481483 room_id: OwnedRoomId,
482484
@@ -508,11 +510,12 @@ impl RoomEventCacheInner {
508510 weak_room: WeakRoom,
509511 state: LockedRoomEventCacheState,
510512 shared_pagination_status: SharedObservable<SharedPaginationStatus>,
513+ generation: Arc<AtomicU64>,
511514 auto_shrink_sender: mpsc::Sender<AutoShrinkChannelPayload>,
512515 update_sender: RoomEventCacheUpdateSender,
513516 ) -> Self {
514517 Self {
515- generation: AtomicU64::new(0),
518+ generation,
516519 room_id,
517520 weak_room,
518521 state,
@@ -1292,9 +1295,8 @@ mod timed_tests {
12921295 assert_eq!(received_room_id, room_id);
12931296 assert!(generic_stream.is_empty());
12941297
1295- // Events individually are not forgotten by the event cache, after clearing a
1296- // room.
1297- assert!(room_event_cache.find_event(event_id1).await.unwrap().is_some());
1298+ // Clearing a room also removes its standalone cached events.
1299+ assert!(room_event_cache.find_event(event_id1).await.unwrap().is_none());
12981300
12991301 // But their presence in a linked chunk is forgotten.
13001302 let items = room_event_cache.events().await.unwrap();
crates/matrix-sdk/src/event_cache/caches/room/reseed.md+2−1
@@ -9,7 +9,8 @@ messages API before reseeding. Duplicate event IDs within the supplied window
99 are retained once. `None` means the start of accessible history has been reached,
1010 including for an empty window. An empty window with a cursor remains pageable.
1111
12-This replaces cached history and invalidates loaded thread timelines. It does
12+This replaces cached history and invalidates thread, pinned, and event-focused
13+timelines, including persisted views that have not been loaded. It does
1314 not establish continuity with discarded history. Account credentials, encryption
1415 state, and pending outgoing messages are not cleared.
1516
crates/matrix-sdk/src/event_cache/caches/room/state.rs+26−14
@@ -16,11 +16,11 @@ use std::{
1616 collections::{BTreeMap, HashMap, HashSet},
1717 sync::{
1818 Arc, OnceLock,
19- atomic::{AtomicUsize, Ordering},
19+ atomic::{AtomicU64, AtomicUsize, Ordering},
2020 },
2121 };
2222
23-use eyeball::SharedObservable;
23+use eyeball::{ObservableWriteGuard, SharedObservable};
2424 use eyeball_im::VectorDiff;
2525 use matrix_sdk_base::{
2626 RoomInfoNotableUpdateReasons, apply_redaction, check_validity_of_replacement_events,
@@ -124,6 +124,7 @@ pub struct RoomEventCacheState {
124124 pinned_event_cache: OnceLock<PinnedEventCache>,
125125
126126 pagination_status: SharedObservable<SharedPaginationStatus>,
127+ generation: Arc<AtomicU64>,
127128
128129 /// A clone of [`super::RoomEventCacheInner::update_sender`].
129130 ///
@@ -310,6 +311,7 @@ impl LockedRoomEventCacheState {
310311 linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
311312 store: EventCacheStoreLock,
312313 pagination_status: SharedObservable<SharedPaginationStatus>,
314+ generation: Arc<AtomicU64>,
313315 automatic_pagination: Option<AutomaticPagination>,
314316 ) -> Result<Self, EventCacheError> {
315317 let store_guard = match store.lock().await? {
@@ -387,6 +389,7 @@ impl LockedRoomEventCacheState {
387389 // permalink.
388390 event_focused_caches: HashMap::new(),
389391 pagination_status,
392+ generation,
390393 update_sender,
391394 linked_chunk_update_sender,
392395 room_version_rules,
@@ -787,10 +790,23 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
787790 let _ = replacement.updates_as_vector_diffs();
788791 self.state.room_linked_chunk = replacement;
789792 self.state.waited_for_initial_prev_token = true;
793+ self.reset_pagination_status(prev_batch.is_none());
790794 Ok(())
791795 }
792796
793- pub async fn clear_loaded_threads(&mut self) -> Result<(), EventCacheError> {
797+ fn reset_pagination_status(&self, hit_timeline_start: bool) {
798+ let mut status = self.state.pagination_status.write();
799+ self.state.generation.fetch_add(1, Ordering::SeqCst);
800+ ObservableWriteGuard::set(&mut status, SharedPaginationStatus::Idle { hit_timeline_start });
801+ }
802+
803+ pub async fn clear_loaded_views(&mut self) -> Result<(), EventCacheError> {
804+ for (_, focused) in self.state.event_focused_caches.drain() {
805+ focused.clear();
806+ }
807+ if let Some(pinned) = self.state.pinned_event_cache.take() {
808+ pinned.clear().await?;
809+ }
794810 for thread in self.state.threads.values_mut() {
795811 thread.clear().await?;
796812 }
@@ -798,28 +814,24 @@ impl<'a> RoomEventCacheStateLockWriteGuard<'a> {
798814 }
799815
800816 async fn reset_internal(&mut self) -> Result<(), EventCacheError> {
801- self.state.room_linked_chunk.reset();
817+ let mut replacement = EventLinkedChunk::new();
818+ replacement.reset();
819+ self.send_updates_to_store(replacement.store_updates().take()).await?;
820+ self.state.room_linked_chunk = replacement;
821+ // Note: this may cancel an ongoing pagination.
822+ self.reset_pagination_status(false);
802823
803824 // No need to update the thread summaries: the room events are
804825 // gone because of the reset of `room_linked_chunk`.
805826 //
806827 // Clear the threads.
807- for thread in self.state.threads.values_mut() {
808- thread.clear().await?;
809- }
810-
811- self.propagate_changes().await?;
828+ self.clear_loaded_views().await?;
812829
813830 // Reset the pagination state too: pretend we never waited for the initial
814831 // prev-batch token, and indicate that we're not at the start of the
815832 // timeline, since we don't know about that anymore.
816833 self.state.waited_for_initial_prev_token = false;
817834
818- // Note: this may cancel an ongoing pagination.
819- self.state
820- .pagination_status
821- .set(SharedPaginationStatus::Idle { hit_timeline_start: false });
822-
823835 Ok(())
824836 }
825837
crates/matrix-sdk/src/event_cache/caches/thread/mod.rs+27−2
@@ -17,8 +17,15 @@
1717 pub mod pagination;
1818 mod state;
1919
20-use std::{fmt, sync::Arc};
20+use std::{
21+ fmt,
22+ sync::{
23+ Arc,
24+ atomic::{AtomicU64, Ordering},
25+ },
26+};
2127
28+use eyeball::{ObservableWriteGuard, SharedObservable};
2229 use matrix_sdk_base::event_cache::{Event, store::EventCacheStoreLock};
2330 use ruma::{EventId, OwnedEventId, OwnedRoomId};
2431 pub(super) use state::LockedThreadEventCacheState;
@@ -27,7 +34,9 @@ use tracing::error;
2734
2835 use self::pagination::ThreadPagination;
2936 use super::{
30- super::Result, EventsOrigin, TimelineVectorDiffs, room::RoomEventCacheLinkedChunkUpdate,
37+ super::{Result, caches::pagination::SharedPaginationStatus},
38+ EventsOrigin, TimelineVectorDiffs,
39+ room::RoomEventCacheLinkedChunkUpdate,
3140 };
3241 use crate::room::WeakRoom;
3342
@@ -38,6 +47,8 @@ pub(super) struct ThreadEventCache {
3847
3948 /// The (non-cloneable) details of the `RoomEventCache`.
4049 struct ThreadEventCacheInner {
50+ generation: AtomicU64,
51+
4152 /// The thread root ID.
4253 thread_id: OwnedEventId,
4354
@@ -46,6 +57,8 @@ struct ThreadEventCacheInner {
4657
4758 /// State for this thread's event cache.
4859 state: LockedThreadEventCacheState,
60+
61+ pagination_status: SharedObservable<SharedPaginationStatus>,
4962 }
5063
5164 impl fmt::Debug for ThreadEventCache {
@@ -65,6 +78,7 @@ impl ThreadEventCache {
6578 ) -> Self {
6679 Self {
6780 inner: Arc::new(ThreadEventCacheInner {
81+ generation: AtomicU64::new(0),
6882 thread_id: thread_id.clone(),
6983 weak_room,
7084 state: LockedThreadEventCacheState::new(
@@ -73,6 +87,9 @@ impl ThreadEventCache {
7387 store,
7488 linked_chunk_update_sender,
7589 ),
90+ pagination_status: SharedObservable::new(SharedPaginationStatus::Idle {
91+ hit_timeline_start: false,
92+ }),
7693 }),
7794 }
7895 }
@@ -100,6 +117,14 @@ impl ThreadEventCache {
100117 let mut state = self.inner.state.write().await?;
101118
102119 let updates_as_vector_diffs = state.reset().await?;
120+ {
121+ let mut status = self.inner.pagination_status.write();
122+ self.inner.generation.fetch_add(1, Ordering::SeqCst);
123+ ObservableWriteGuard::set(
124+ &mut status,
125+ SharedPaginationStatus::Idle { hit_timeline_start: false },
126+ );
127+ }
103128
104129 if !updates_as_vector_diffs.is_empty() {
105130 let _ = state.state.sender.send(TimelineVectorDiffs {
crates/matrix-sdk/src/event_cache/caches/thread/pagination.rs+15−12
@@ -42,10 +42,6 @@ use crate::{
4242 #[derive(Clone)]
4343 struct ThreadEventCacheWrapper {
4444 cache: Arc<ThreadEventCacheInner>,
45-
46- // Threads do not support pagination status for the moment but we need one, so let's use a
47- // dummy one for now.
48- dummy_pagination_status: SharedObservable<SharedPaginationStatus>,
4945 }
5046
5147 /// An API object to run pagination queries on a `ThreadEventCache`.
@@ -55,12 +51,7 @@ pub struct ThreadPagination(Pagination<ThreadEventCacheWrapper>);
5551 impl ThreadPagination {
5652 /// Construct a new [`ThreadPagination`].
5753 pub(super) fn new(cache: Arc<ThreadEventCacheInner>) -> Self {
58- Self(Pagination::new(ThreadEventCacheWrapper {
59- cache,
60- dummy_pagination_status: SharedObservable::new(SharedPaginationStatus::Idle {
61- hit_timeline_start: false,
62- }),
63- }))
54+ Self(Pagination::new(ThreadEventCacheWrapper { cache }))
6455 }
6556
6657 /// Starts a back-pagination for the requested number of events.
@@ -91,7 +82,11 @@ impl ThreadPagination {
9182
9283 impl PaginatedCache for ThreadEventCacheWrapper {
9384 fn status(&self) -> &SharedObservable<SharedPaginationStatus> {
94- &self.dummy_pagination_status
85+ &self.cache.pagination_status
86+ }
87+
88+ fn generation(&self) -> u64 {
89+ self.cache.generation.load(std::sync::atomic::Ordering::SeqCst)
9590 }
9691
9792 async fn load_more_events_backwards(&self) -> Result<LoadMoreEventsBackwardsOutcome> {
@@ -180,8 +175,12 @@ impl PaginatedCache for ThreadEventCacheWrapper {
180175 mut events: Vec<Event>,
181176 prev_token: Option<String>,
182177 new_token: Option<String>,
183- _generation: u64,
178+ generation: u64,
184179 ) -> Result<Option<BackPaginationOutcome>> {
180+ if self.generation() != generation {
181+ return Ok(None);
182+ }
183+
185184 let Some(room) = self.cache.weak_room.get() else {
186185 // The client is shutting down.
187186 return Ok(None);
@@ -204,6 +203,10 @@ impl PaginatedCache for ThreadEventCacheWrapper {
204203
205204 let mut state = self.cache.state.write().await?;
206205
206+ if self.generation() != generation {
207+ return Ok(None);
208+ }
209+
207210 // Save all the events (but the thread root) in the store.
208211 state.save_events(events.iter().cloned()).await?;
209212
crates/matrix-sdk/tests/history_reseed.rs+339−3
@@ -5,15 +5,19 @@ use std::{
55
66 use eyeball_im::VectorDiff;
77 use matrix_sdk::{
8+ ThreadingSupport,
89 config::StoreConfig,
9- event_cache::{PaginationStatus, RoomEventCacheUpdate},
10+ event_cache::{EventFocusThreadMode, PaginationStatus, RoomEventCacheUpdate},
1011 room::MessagesOptions,
1112 ruma::{event_id, room_id},
12- test_utils::mocks::{MatrixMockServer, RoomMessagesResponseTemplate},
13+ test_utils::mocks::{
14+ MatrixMockServer, RoomContextResponseTemplate, RoomMessagesResponseTemplate,
15+ RoomRelationsResponseTemplate,
16+ },
1317 };
1418 use matrix_sdk_base::{event_cache::store::EventCacheStore, linked_chunk::LinkedChunkId};
1519 use matrix_sdk_common::cross_process_lock::CrossProcessLockConfig;
16-use matrix_sdk_test::{ALICE, event_factory::EventFactory};
20+use matrix_sdk_test::{ALICE, JoinedRoomBuilder, event_factory::EventFactory};
1721
1822 #[path = "history_reseed/failing_store.rs"]
1923 mod failing_store;
@@ -210,6 +214,338 @@ async fn reseed_preserves_timeline_when_store_write_fails() {
210214 assert_eq!(format!("{current_chunks:?}"), format!("{original_chunks:?}"));
211215 }
212216
217+#[tokio::test]
218+async fn reseed_redaction_does_not_leak_stale_diff_into_next_sync() {
219+ let server = MatrixMockServer::new().await;
220+ let client = server.client_builder().build().await;
221+ let room_id = room_id!("!reseed:localhost");
222+ client.event_cache().subscribe().unwrap();
223+ server.sync_joined_room(&client, room_id).await;
224+ let room = client.get_room(room_id).unwrap();
225+ let (cache, _handles) = room.event_cache().await.unwrap();
226+ let (_, mut updates) = cache.subscribe().await.unwrap();
227+ let factory = EventFactory::new().room(room_id).sender(&ALICE);
228+ let redacted_id = event_id!("$redacted:localhost");
229+ let redaction_id = event_id!("$redaction:localhost");
230+
231+ cache
232+ .reseed(
233+ vec![
234+ factory.text_msg("secret").event_id(redacted_id).into_event(),
235+ factory.redaction(&redacted_id).event_id(redaction_id).into_event(),
236+ ],
237+ None,
238+ )
239+ .await
240+ .unwrap();
241+ let reset =
242+ tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
243+ let RoomEventCacheUpdate::UpdateTimelineEvents(reset) = reset else {
244+ panic!("expected timeline reset")
245+ };
246+ let VectorDiff::Reset { values } = &reset.diffs[0] else { panic!("expected reset diff") };
247+ assert_eq!(values.len(), 2);
248+ assert!(values[0].raw().deserialize().unwrap().is_redacted());
249+
250+ let new_event_id = event_id!("$new-live:localhost");
251+ server
252+ .sync_room(
253+ &client,
254+ JoinedRoomBuilder::new(room_id).add_timeline_event(
255+ factory.text_msg("live").event_id(new_event_id).into_raw_sync(),
256+ ),
257+ )
258+ .await;
259+ let live_update =
260+ tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
261+ let RoomEventCacheUpdate::UpdateTimelineEvents(live_update) = live_update else {
262+ panic!("expected timeline update")
263+ };
264+ assert!(live_update.diffs.iter().all(|diff| !matches!(diff, VectorDiff::Set { .. })));
265+ assert!(live_update.diffs.iter().all(|diff| match diff {
266+ VectorDiff::Append { values } => {
267+ values.iter().all(|event| event.event_id().as_deref() == Some(new_event_id))
268+ }
269+ VectorDiff::PushBack { value } => value.event_id().as_deref() == Some(new_event_id),
270+ _ => true,
271+ }));
272+ assert_eq!(cache.events().await.unwrap().len(), 3);
273+}
274+
275+#[tokio::test]
276+async fn reseed_retries_thread_pagination_without_stale_response() {
277+ use wiremock::ResponseTemplate;
278+
279+ let server = MatrixMockServer::new().await;
280+ let client = server
281+ .client_builder()
282+ .on_builder(|builder| {
283+ builder.with_threading_support(ThreadingSupport::Enabled { with_subscriptions: false })
284+ })
285+ .build()
286+ .await;
287+ let room_id = room_id!("!reseed:localhost");
288+ client.event_cache().subscribe().unwrap();
289+ let thread_root_id = event_id!("$thread-root:localhost");
290+ let initial_reply_id = event_id!("$initial-reply:localhost");
291+ let factory = EventFactory::new().room(room_id).sender(&ALICE);
292+ let room = server
293+ .sync_room(
294+ &client,
295+ JoinedRoomBuilder::new(room_id)
296+ .set_timeline_prev_batch("initial-thread-cursor")
297+ .add_timeline_event(
298+ factory
299+ .text_msg("initial reply")
300+ .in_thread(thread_root_id, thread_root_id)
301+ .event_id(initial_reply_id),
302+ ),
303+ )
304+ .await;
305+ let (cache, _handles) = room.event_cache().await.unwrap();
306+ let (initial_events, _thread_updates) =
307+ cache.subscribe_to_thread(thread_root_id.to_owned()).await.unwrap();
308+ assert_eq!(initial_events.len(), 1);
309+ assert_eq!(initial_events[0].event_id().as_deref(), Some(initial_reply_id));
310+
311+ let stale_reply_id = event_id!("$stale-reply:localhost");
312+ let fresh_reply_id = event_id!("$fresh-reply:localhost");
313+ server
314+ .mock_room_relations()
315+ .match_target_event(thread_root_id.to_owned())
316+ .ok(RoomRelationsResponseTemplate::default().events(vec![
317+ factory
318+ .text_msg("stale reply")
319+ .in_thread(thread_root_id, thread_root_id)
320+ .event_id(stale_reply_id),
321+ ]))
322+ .mock_once()
323+ .mount()
324+ .await;
325+ let root_event = factory.text_msg("thread root").event_id(thread_root_id).into_event();
326+ server
327+ .mock_room_event()
328+ .room(room_id)
329+ .ok_with_template(
330+ ResponseTemplate::new(200)
331+ .set_body_json(root_event.into_raw().json())
332+ .set_delay(Duration::from_millis(300)),
333+ )
334+ .mock_once()
335+ .mount()
336+ .await;
337+
338+ let pagination = tokio::spawn({
339+ let cache = cache.clone();
340+ let thread_root_id = thread_root_id.to_owned();
341+ async move {
342+ cache.thread_pagination(thread_root_id).await.unwrap().run_backwards_once(10).await
343+ }
344+ });
345+ tokio::time::timeout(Duration::from_secs(8), async {
346+ loop {
347+ if server
348+ .received_requests()
349+ .await
350+ .unwrap()
351+ .iter()
352+ .any(|request| request.url.path().contains("/event/"))
353+ {
354+ break;
355+ }
356+ tokio::task::yield_now().await;
357+ }
358+ })
359+ .await
360+ .unwrap();
361+ server
362+ .mock_room_relations()
363+ .match_target_event(thread_root_id.to_owned())
364+ .ok(RoomRelationsResponseTemplate::default()
365+ .events(vec![
366+ factory
367+ .text_msg("fresh reply")
368+ .in_thread(thread_root_id, thread_root_id)
369+ .event_id(fresh_reply_id),
370+ ])
371+ .next_batch("fresh-cursor"))
372+ .mock_once()
373+ .mount()
374+ .await;
375+ cache.reseed(Vec::new(), None).await.unwrap();
376+
377+ let outcome =
378+ tokio::time::timeout(Duration::from_secs(8), pagination).await.unwrap().unwrap().unwrap();
379+ assert!(outcome.events.iter().all(|event| event.event_id().as_deref() != Some(stale_reply_id)));
380+ assert!(outcome.events.iter().any(|event| event.event_id().as_deref() == Some(fresh_reply_id)));
381+ let (thread_events, _thread_updates) =
382+ cache.subscribe_to_thread(thread_root_id.to_owned()).await.unwrap();
383+ assert!(thread_events.iter().all(|event| event.event_id().as_deref() != Some(stale_reply_id)));
384+ assert!(thread_events.iter().any(|event| event.event_id().as_deref() == Some(fresh_reply_id)));
385+}
386+
387+#[tokio::test]
388+async fn clear_invalidates_focused_cache_clones_and_inflight_pagination() {
389+ let server = MatrixMockServer::new().await;
390+ let client = server.client_builder().build().await;
391+ let room_id = room_id!("!reseed:localhost");
392+ client.event_cache().subscribe().unwrap();
393+ server.sync_joined_room(&client, room_id).await;
394+ let room = client.get_room(room_id).unwrap();
395+ let (cache, _handles) = room.event_cache().await.unwrap();
396+ let factory = EventFactory::new().room(room_id).sender(&ALICE);
397+ let focused_id = event_id!("$focused:localhost");
398+ server
399+ .mock_room_event_context()
400+ .ok(RoomContextResponseTemplate::new(
401+ factory.text_msg("focused").event_id(focused_id).into_event(),
402+ )
403+ .start("old-focused-cursor"))
404+ .mock_once()
405+ .mount()
406+ .await;
407+ let focused = cache
408+ .get_or_create_event_focused_cache(
409+ focused_id.to_owned(),
410+ 10,
411+ EventFocusThreadMode::Automatic,
412+ )
413+ .await
414+ .unwrap();
415+ let focused_clone = focused.clone();
416+ let (initial_events, mut updates) = focused.subscribe().await;
417+ assert_eq!(initial_events.len(), 1);
418+ assert_eq!(initial_events[0].event_id().as_deref(), Some(focused_id));
419+
420+ let stale_event_id = event_id!("$stale-focused:localhost");
421+ server
422+ .mock_room_messages()
423+ .match_from("old-focused-cursor")
424+ .ok(RoomMessagesResponseTemplate::default()
425+ .events(vec![factory.text_msg("stale").event_id(stale_event_id)])
426+ .end_token("stale-next-cursor")
427+ .with_delay(Duration::from_millis(300)))
428+ .mock_once()
429+ .mount()
430+ .await;
431+ let pagination = tokio::spawn({
432+ let focused = focused.clone();
433+ async move { focused.paginate_backwards(10).await }
434+ });
435+ tokio::time::timeout(Duration::from_secs(8), async {
436+ loop {
437+ if server
438+ .received_requests()
439+ .await
440+ .unwrap()
441+ .iter()
442+ .any(|request| request.url.path().ends_with("/messages"))
443+ {
444+ break;
445+ }
446+ tokio::task::yield_now().await;
447+ }
448+ })
449+ .await
450+ .unwrap();
451+
452+ tokio::time::timeout(Duration::from_secs(2), cache.clear()).await.unwrap().unwrap();
453+ let clear =
454+ tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
455+ assert!(matches!(clear.diffs.as_slice(), [VectorDiff::Clear]));
456+ tokio::time::timeout(Duration::from_secs(3), pagination).await.unwrap().unwrap().unwrap();
457+ let (cloned_events, _cloned_updates) = focused_clone.subscribe().await;
458+ assert!(cloned_events.is_empty());
459+ let result = focused_clone.paginate_backwards(10).await.unwrap();
460+ assert!(result.hit_end_of_timeline);
461+ let (events_after_pagination, _updates_after_pagination) = focused_clone.subscribe().await;
462+ assert!(events_after_pagination.is_empty());
463+}
464+
465+#[tokio::test]
466+async fn clear_empties_subscribed_thread_cache() {
467+ let server = MatrixMockServer::new().await;
468+ let client = server
469+ .client_builder()
470+ .on_builder(|builder| {
471+ builder.with_threading_support(ThreadingSupport::Enabled { with_subscriptions: false })
472+ })
473+ .build()
474+ .await;
475+ let room_id = room_id!("!reseed:localhost");
476+ client.event_cache().subscribe().unwrap();
477+ let thread_root_id = event_id!("$thread-root:localhost");
478+ let thread_reply_id = event_id!("$thread-reply:localhost");
479+ let factory = EventFactory::new().room(room_id).sender(&ALICE);
480+ let room = server
481+ .sync_room(
482+ &client,
483+ JoinedRoomBuilder::new(room_id).add_timeline_event(
484+ factory
485+ .text_msg("thread reply")
486+ .in_thread(thread_root_id, thread_root_id)
487+ .event_id(thread_reply_id),
488+ ),
489+ )
490+ .await;
491+ let (cache, _handles) = room.event_cache().await.unwrap();
492+ let (initial_events, mut updates) =
493+ cache.subscribe_to_thread(thread_root_id.to_owned()).await.unwrap();
494+ assert_eq!(initial_events.len(), 1);
495+ assert_eq!(initial_events[0].event_id().as_deref(), Some(thread_reply_id));
496+
497+ cache.clear().await.unwrap();
498+ let clear =
499+ tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
500+ assert!(matches!(clear.diffs.as_slice(), [VectorDiff::Clear]));
501+ let (events_after_clear, _updates_after_clear) =
502+ cache.subscribe_to_thread(thread_root_id.to_owned()).await.unwrap();
503+ assert!(events_after_clear.is_empty());
504+}
505+
506+#[tokio::test]
507+async fn clear_empties_loaded_pinned_cache() {
508+ let server = MatrixMockServer::new().await;
509+ let room_id = room_id!("!reseed:localhost");
510+ let pinned_event_id = event_id!("$pinned:localhost");
511+ let factory = EventFactory::new().room(room_id).sender(&ALICE);
512+ let pinned_event = factory.text_msg("pinned").event_id(pinned_event_id).into_event();
513+ server.mock_room_event().match_event_id().ok(pinned_event).mock_once().mount().await;
514+ let client = server.client_builder().build().await;
515+ client.event_cache().subscribe().unwrap();
516+ let room = server
517+ .sync_room(
518+ &client,
519+ JoinedRoomBuilder::new(room_id).add_state_bulk(vec![
520+ factory.room_pinned_events(vec![pinned_event_id.to_owned()]).into(),
521+ ]),
522+ )
523+ .await;
524+ let (cache, _handles) = room.event_cache().await.unwrap();
525+ let (initial_events, mut updates) = cache.subscribe_to_pinned_events().await.unwrap();
526+ let mut visible: imbl::Vector<_> = initial_events.into();
527+ tokio::time::timeout(Duration::from_secs(3), async {
528+ while visible.is_empty() {
529+ for diff in updates.recv().await.unwrap().diffs {
530+ diff.apply(&mut visible);
531+ }
532+ }
533+ })
534+ .await
535+ .unwrap();
536+ assert_eq!(visible.len(), 1);
537+ assert_eq!(visible[0].event_id().as_deref(), Some(pinned_event_id));
538+
539+ cache.clear().await.unwrap();
540+ let clear =
541+ tokio::time::timeout(Duration::from_secs(2), updates.recv()).await.unwrap().unwrap();
542+ assert!(matches!(clear.diffs.as_slice(), [VectorDiff::Clear]));
543+ for diff in clear.diffs {
544+ diff.apply(&mut visible);
545+ }
546+ assert!(visible.is_empty());
547+}
548+
213549 #[cfg(feature = "sqlite")]
214550 #[tokio::test]
215551 async fn reseed_persists_the_window_and_cursor_across_client_recreation() {