Dima/matrix-rust-sdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Allow storing the same Event in multiple LinkedChunks of the same Room (#6200)

Browse Source 
# Overview

There are scenarios in which it is sensible to have an event exist in
the same room more than once. Notably, this is true in the context of a
thread, where an event exists in the main timeline of a room, as well as
in a thread of that same room.

Support for this behavior has been implemented in the
`SQLiteEventCacheStore` in #6065; however, this was never implemented
for the `IndexeddbEventCacheStore` or the `MemoryStore`. This pull
request extends this behavior to both of those stores.

# Changes

## Integration Tests
First, `test_event_chunks_allows_same_event_in_room_and_thread` was
moved from `matrix_sdk_sqlite::event_cache_store` to
`matrix_sdk_base::event_cache::store::integration_tests`. Then, a few
additional integration tests were added to ensure that behavior is
consistent across implementations of `EventCacheStore`.

## `IndexeddbEventCacheStore`
In order to accommodate the behavioral changes specified by the
integration tests, it was necessary to modify the schema in the
IndexedDB implementation of `EventCacheStore`. Namely, the events object
store was cleared and removed and then replaced with a nearly identical
one, the only difference being the removal of a uniqueness constraint on
one of the indices.

The remaining changes mostly involved updating the behavior of top-level
`EventCacheStore` functions - e.g., filtering out events where they were
duplicated or removing positioning information where it was not
relevant.

## `MemoryStore`
The changes to `MemoryStore` mostly involved updating the behavior of
top-level `EventCacheStore` function - e.g., filtering out events where
they were duplicated or removing positioning information where it was
not relevant.

That being said, it also involved some breaking changes to
`RelationalLinkedChunk`.

1. `RelationalLinkedChunk::items` - this function returned an `Iterator`
that did not contain information about the `LinkedChunkId`, so this
information was added to the items in the `Iterator`.
2. `RelationalLinkedChunk::save_item` - this function did not update the
item in all linked chunks of the provided `Room`. It now does this, but
requires that the provided `Item` be `Clone`.

(1) could probably have been a new function, but I thought a nicer
interface was worth the breaking change. (2) could probably be prevented
by re-organizing `RelationalLinkedChunk`'s internal data structures to
remove the `Clone` requirement, but that seemed like it could turn into
a large refactoring project, so I opted for something simpler albeit
somewhat crude.

In both cases, I'm open to suggestions and would be happy to revisit if
something else is preferred.

---
Closes #6094.

- [x] I've documented the public API Changes in the appropriate
`CHANGELOG.md` files.
- [x] I've read [the `CONTRIBUTING.md`
file](https://github.com/matrix-org/matrix-rust-sdk/blob/main/CONTRIBUTING.md),
notably the sections about Pull requests, Commit message format, and AI
policy.
- [ ] This PR was made with the help of AI.

Signed-off-by: Michael Goldenberg <m@mgoldenberg.net>

---------

Signed-off-by: Michael Goldenberg <m@mgoldenberg.net>
This commit is contained in:
mgoldenberg
2026-03-10 09:55:02 -04:00
committed by GitHub
parent b65e450813
commit 4ec9124ce1
13 changed files with 689 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/matrix-sdk-base/CHANGELOG.md
+5
View File
@@ -25,6 +25,11 @@ All notable changes to this project will be documented in this file.
### Features
- Add support in the `MemoryStore`'s implementation of `EventCacheStore` for
having duplicate events in a room, where each duplicate is in a different
`LinkedChunk`. This is useful, e.g., when an event is in a room and a
thread in that room.
(#[6200](https://github.com/matrix-org/matrix-rust-sdk/pull/6200))
- Add `StateStore::upsert_thread_subscriptions()` method for bulk upserts.
([#5848](https://github.com/matrix-org/matrix-rust-sdk/pull/5848))
- The `LatestEventValue::LocalHasBeenSent` variant gains a new `event_id:
crates/matrix-sdk-base/src/event_cache/store/integration_tests.rs
+422
View File
@@ -141,6 +141,10 @@ pub trait EventCacheStoreIntegrationTests {
/// already exist in the store.
async fn test_linked_chunk_exists_before_referenced(&self);
/// Test that the same event can exist in a room's linked chunk and a
/// thread's linked chunk simultaneously.
async fn test_linked_chunk_allows_same_event_in_room_and_thread(&self);
/// Test loading the last chunk in a linked chunk from the store.
async fn test_load_last_chunk(&self);
@@ -202,18 +206,34 @@ pub trait EventCacheStoreIntegrationTests {
/// Test that an event can be found or not.
async fn test_find_event(&self);
/// Test that an event can be found when it exists in both a room and a
/// thread in that room.
async fn test_find_event_when_event_in_room_and_thread(&self);
/// Test that finding event relations works as expected.
async fn test_find_event_relations(&self);
/// Test that find event relations works as expected when an event is both a
/// room and a thread in that room.
async fn test_find_event_relations_when_event_in_room_and_thread(&self);
/// Test that getting all events in a room works as expected.
async fn test_get_room_events(&self);
/// Test that getting events in a room of a certain type works as expected.
async fn test_get_room_events_filtered(&self);
/// Test that getting all events in a room works as expected when the event
/// is in both a room and thread in that room.
async fn test_get_room_events_with_event_in_room_and_thread(&self);
/// Test that saving an event works as expected.
async fn test_save_event(&self);
/// Test that saving an existing event updates it's contents in both room
/// and thread linked chunks.
async fn test_save_event_updates_event_in_room_and_thread(&self);
/// Test multiple things related to distinguishing a thread linked chunk
/// from a room linked chunk.
async fn test_thread_vs_room_linked_chunk(&self);
@@ -355,6 +375,62 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
.unwrap_err();
}
async fn test_linked_chunk_allows_same_event_in_room_and_thread(&self) {
// This test verifies that the same event can appear in both a room's linked
// chunk and a thread's linked chunk. This is the real-world use case:
// a thread reply appears in both the main room timeline and the thread.
let room_id = *DEFAULT_TEST_ROOM_ID;
let thread_root = event_id!("$thread_root");
// Create an event that will be inserted into both the room and thread linked
// chunks.
let event_id = event_id!("$thread_reply");
let event = make_test_event_with_event_id(room_id, "thread reply", Some(event_id));
let room_linked_chunk_id = LinkedChunkId::Room(room_id);
let thread_linked_chunk_id = LinkedChunkId::Thread(room_id, thread_root);
// Insert the event into the room's linked chunk.
self.handle_linked_chunk_updates(
room_linked_chunk_id,
vec![
Update::NewItemsChunk { previous: None, new: CId::new(1), next: None },
Update::PushItems { at: Position::new(CId::new(1), 0), items: vec![event.clone()] },
],
)
.await
.unwrap();
// Insert the same event into the thread's linked chunk.
self.handle_linked_chunk_updates(
thread_linked_chunk_id,
vec![
Update::NewItemsChunk { previous: None, new: CId::new(1), next: None },
Update::PushItems { at: Position::new(CId::new(1), 0), items: vec![event] },
],
)
.await
.unwrap();
// Verify both entries exist by loading chunks from both linked chunk IDs.
let room_chunks = self.load_all_chunks(room_linked_chunk_id).await.unwrap();
let thread_chunks = self.load_all_chunks(thread_linked_chunk_id).await.unwrap();
assert_eq!(room_chunks.len(), 1);
assert_eq!(thread_chunks.len(), 1);
// Verify the event is in both.
assert_matches!(&room_chunks[0].content, ChunkContent::Items(events) => {
assert_eq!(events.len(), 1);
assert_eq!(events[0].event_id().as_deref(), Some(event_id));
});
assert_matches!(&thread_chunks[0].content, ChunkContent::Items(events) => {
assert_eq!(events.len(), 1);
assert_eq!(events[0].event_id().as_deref(), Some(event_id));
});
}
async fn test_load_all_chunks_metadata(&self) {
let room_id = room_id!("!r0:matrix.org");
let linked_chunk_id = LinkedChunkId::Room(room_id);
@@ -1460,6 +1536,75 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
);
}
async fn test_find_event_when_event_in_room_and_thread(&self) {
let room_id = *DEFAULT_TEST_ROOM_ID;
let thread_root = event_id!("$thread_root");
// Create an event that will be only be inserted into the room
let room_event_id = event_id!("$room_event");
let room_event = make_test_event_with_event_id(room_id, "room event", Some(room_event_id));
// Create an event that will only be inserted into the thread
let thread_event_id = event_id!("$thread_event");
let thread_event =
make_test_event_with_event_id(room_id, "thread event", Some(thread_event_id));
// Create an event that will be inserted into both the room and thread linked
// chunks.
let room_and_thread_event_id = event_id!("$room_and_thread");
let room_and_thread_event = make_test_event_with_event_id(
room_id,
"room and thread",
Some(room_and_thread_event_id),
);
let room_linked_chunk_id = LinkedChunkId::Room(room_id);
let thread_linked_chunk_id = LinkedChunkId::Thread(room_id, thread_root);
// Insert the relevant events into the room's linked chunk.
self.handle_linked_chunk_updates(
room_linked_chunk_id,
vec![
Update::NewItemsChunk { previous: None, new: CId::new(1), next: None },
Update::PushItems {
at: Position::new(CId::new(1), 0),
items: vec![room_event, room_and_thread_event.clone()],
},
],
)
.await
.unwrap();
// Insert the relevant events into the thread's linked chunk.
self.handle_linked_chunk_updates(
thread_linked_chunk_id,
vec![
Update::NewItemsChunk { previous: None, new: CId::new(1), next: None },
Update::PushItems {
at: Position::new(CId::new(1), 0),
items: vec![thread_event, room_and_thread_event],
},
],
)
.await
.unwrap();
// Verify that event that is only in the room can be retrieved
assert_matches!(self.find_event(room_id, room_event_id).await, Ok(Some(event)) => {
assert_eq!(event.event_id().unwrap(), room_event_id)
});
// Verify that the event that is only in the thread can be retrieved
assert_matches!(self.find_event(room_id, thread_event_id).await, Ok(Some(event)) => {
assert_eq!(event.event_id().unwrap(), thread_event_id)
});
// Verify that event that is in both room and thread can be retrieved
assert_matches!(self.find_event(room_id, room_and_thread_event_id).await, Ok(Some(event)) => {
assert_eq!(event.event_id().unwrap(), room_and_thread_event_id);
});
}
async fn test_find_event_relations(&self) {
let room_id = room_id!("!r0:matrix.org");
let another_room_id = room_id!("!r1:matrix.org");
@@ -1574,6 +1719,121 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
);
}
async fn test_find_event_relations_when_event_in_room_and_thread(&self) {
let room_id = *DEFAULT_TEST_ROOM_ID;
let thread_root = event_id!("$thread_root");
// Create an event that will inserted into both the room and thread linked
// chunks.
let event_id = event_id!("$event");
let event = make_test_event_with_event_id(room_id, "event", Some(event_id));
// Create an event that will only be inserted into the thread in order to help
// distinguish between the room and thread linked chunks.
let extra_thread_event_id = event_id!("$extra_thread_event");
let extra_thread_event = make_test_event_with_event_id(
room_id,
"extra thread event",
Some(extra_thread_event_id),
);
// Create a reaction that will only be inserted into the room
let room_reaction_id = event_id!("$room_reaction");
let room_reaction = EventFactory::new()
.room(room_id)
.sender(*ALICE)
.reaction(event_id, "room")
.event_id(room_reaction_id)
.into_event();
// Create a reaction that will only be inserted into the thread
let thread_reaction_id = event_id!("$thread_reaction");
let thread_reaction = EventFactory::new()
.room(room_id)
.sender(*ALICE)
.reaction(event_id, "thread")
.event_id(thread_reaction_id)
.into_event();
// Create a reaction that will be inserted into both the room and thread linked
// chunks.
let room_and_thread_reaction_id = event_id!("$room_and_thread_reaction");
let room_and_thread_reaction = EventFactory::new()
.room(room_id)
.sender(*ALICE)
.reaction(event_id, "room and thread")
.event_id(room_and_thread_reaction_id)
.into_event();
let room_linked_chunk_id = LinkedChunkId::Room(room_id);
let thread_linked_chunk_id = LinkedChunkId::Thread(room_id, thread_root);
// Insert the relevant events into the room's linked chunk.
self.handle_linked_chunk_updates(
room_linked_chunk_id,
vec![
Update::NewItemsChunk { previous: None, new: CId::new(1), next: None },
Update::PushItems {
at: Position::new(CId::new(1), 0),
items: vec![event.clone(), room_reaction, room_and_thread_reaction.clone()],
},
],
)
.await
.unwrap();
// Insert the relevant events into the thread's linked chunk.
self.handle_linked_chunk_updates(
thread_linked_chunk_id,
vec![
Update::NewItemsChunk { previous: None, new: CId::new(1), next: None },
Update::PushItems {
at: Position::new(CId::new(1), 0),
items: vec![
event.clone(),
extra_thread_event,
thread_reaction,
room_and_thread_reaction,
],
},
],
)
.await
.unwrap();
// Verify that only related events from the room are returned
assert_matches!(self.find_event_relations(room_id, event_id, None).await, Ok(relations) => {
assert_eq!(relations.len(), 3);
// Verify that room reaction is in the list and associated with its
// position in the room linked chunk.
let room_relation = relations
.iter()
.find(|relation| relation.0.event_id().unwrap() == room_reaction_id)
.unwrap();
assert_matches!(room_relation, (_, Some(position)) => {
assert_eq!(*position, Position::new(CId::new(1), 1));
});
// Verify that thread reaction is in the list and not associated with a
// position, as all positions are provided for the room linked chunk.
let thread_relation = relations
.iter()
.find(|relation| relation.0.event_id().unwrap() == thread_reaction_id)
.unwrap();
assert_matches!(thread_relation, (_, None));
// Verify that room and thread reaction is in the list and associated
// with its position in the room linked chunk, not the thread linked chunk.
let room_and_thread_relation = relations
.iter()
.find(|relation| relation.0.event_id().unwrap() == room_and_thread_reaction_id)
.unwrap();
assert_matches!(room_and_thread_relation, (_, Some(position)) => {
assert_eq!(*position, Position::new(CId::new(1), 2));
});
});
}
async fn test_get_room_events(&self) {
let room_id = room_id!("!r0:matrix.org");
let another_room_id = room_id!("!r1:matrix.org");
@@ -1702,6 +1962,73 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
assert_expected_events!(events, [first_event]);
}
async fn test_get_room_events_with_event_in_room_and_thread(&self) {
let room_id = *DEFAULT_TEST_ROOM_ID;
let thread_root = event_id!("$thread_root");
// Create an event that will be only be inserted into the room
let room_event_id = event_id!("$room_event");
let room_event = make_test_event_with_event_id(room_id, "room event", Some(room_event_id));
// Create an event that will only be inserted into the thread. This may not be a
// sensible operation in practice, as threads seem to always exist in a
// room, but let's test it anyway.
let thread_event_id = event_id!("$thread_event");
let thread_event =
make_test_event_with_event_id(room_id, "thread event", Some(thread_event_id));
// Create an event that will be inserted into both the room and thread linked
// chunks.
let room_and_thread_event_id = event_id!("$room_and_thread");
let room_and_thread_event = make_test_event_with_event_id(
room_id,
"room and thread",
Some(room_and_thread_event_id),
);
let room_linked_chunk_id = LinkedChunkId::Room(room_id);
let thread_linked_chunk_id = LinkedChunkId::Thread(room_id, thread_root);
// Insert the relevant events into the room's linked chunk.
self.handle_linked_chunk_updates(
room_linked_chunk_id,
vec![
Update::NewItemsChunk { previous: None, new: CId::new(1), next: None },
Update::PushItems {
at: Position::new(CId::new(1), 0),
items: vec![room_event, room_and_thread_event.clone()],
},
],
)
.await
.unwrap();
// Insert the relevant events into the thread's linked chunk.
self.handle_linked_chunk_updates(
thread_linked_chunk_id,
vec![
Update::NewItemsChunk { previous: None, new: CId::new(1), next: None },
Update::PushItems {
at: Position::new(CId::new(1), 0),
items: vec![thread_event, room_and_thread_event],
},
],
)
.await
.unwrap();
// Verify that all events can be retrieved and none are duplicated in the
// returned list.
let expected_event_ids =
BTreeSet::from([room_event_id, thread_event_id, room_and_thread_event_id]);
assert_matches!(self.get_room_events(room_id, None, None).await, Ok(events) => {
assert_eq!(events.len(), 3);
assert!(events.iter().all(|event| {
expected_event_ids.contains(&event.event_id().unwrap().as_ref())
}));
});
}
async fn test_save_event(&self) {
let room_id = room_id!("!r0:matrix.org");
let another_room_id = room_id!("!r1:matrix.org");
@@ -1746,6 +2073,66 @@ impl EventCacheStoreIntegrationTests for DynEventCacheStore {
);
}
async fn test_save_event_updates_event_in_room_and_thread(&self) {
let room_id = *DEFAULT_TEST_ROOM_ID;
let thread_root = event_id!("$thread_root");
// Create an event that will be inserted into both the room and thread linked
// chunks.
let event_id = event_id!("$event");
let event = make_test_event_with_event_id(room_id, "event", Some(event_id));
let room_linked_chunk_id = LinkedChunkId::Room(room_id);
let thread_linked_chunk_id = LinkedChunkId::Thread(room_id, thread_root);
// Insert the relevant events into the room's linked chunk.
self.handle_linked_chunk_updates(
room_linked_chunk_id,
vec![
Update::NewItemsChunk { previous: None, new: CId::new(1), next: None },
Update::PushItems { at: Position::new(CId::new(1), 0), items: vec![event.clone()] },
],
)
.await
.unwrap();
// Insert the relevant events into the thread's linked chunk.
self.handle_linked_chunk_updates(
thread_linked_chunk_id,
vec![
Update::NewItemsChunk { previous: None, new: CId::new(1), next: None },
Update::PushItems { at: Position::new(CId::new(1), 0), items: vec![event.clone()] },
],
)
.await
.unwrap();
// Save updated version of original event, which should replace the content of
// the existing event
let updated_content = "updated content";
let updated = make_test_event_with_event_id(room_id, updated_content, Some(event_id));
self.save_event(room_id, updated).await.unwrap();
// Load all chunks from both room and thread
let room_chunks = self.load_all_chunks(room_linked_chunk_id).await.unwrap();
let thread_chunks = self.load_all_chunks(thread_linked_chunk_id).await.unwrap();
assert_eq!(room_chunks.len(), 1);
assert_eq!(thread_chunks.len(), 1);
// Verify the event has been updated in both room and thread
assert_matches!(&room_chunks[0].content, ChunkContent::Items(events) => {
assert_eq!(events.len(), 1);
assert_eq!(events[0].event_id().as_deref(), Some(event_id));
check_test_event(&events[0], updated_content);
});
assert_matches!(&thread_chunks[0].content, ChunkContent::Items(events) => {
assert_eq!(events.len(), 1);
assert_eq!(events[0].event_id().as_deref(), Some(event_id));
check_test_event(&events[0], updated_content);
});
}
async fn test_thread_vs_room_linked_chunk(&self) {
let room_id = room_id!("!r0:matrix.org");
@@ -1930,6 +2317,13 @@ macro_rules! event_cache_store_integration_tests {
event_cache_store.test_linked_chunk_exists_before_referenced().await;
}
#[async_test]
async fn test_linked_chunk_allow_same_event_in_room_and_thread() {
let event_cache_store =
get_event_cache_store().await.unwrap().into_event_cache_store();
event_cache_store.test_linked_chunk_allows_same_event_in_room_and_thread().await;
}
#[async_test]
async fn test_load_last_chunk() {
let event_cache_store =
@@ -2063,6 +2457,13 @@ macro_rules! event_cache_store_integration_tests {
event_cache_store.test_find_event().await;
}
#[async_test]
async fn test_find_event_when_event_in_room_and_thread() {
let event_cache_store =
get_event_cache_store().await.unwrap().into_event_cache_store();
event_cache_store.test_find_event_when_event_in_room_and_thread().await;
}
#[async_test]
async fn test_find_event_relations() {
let event_cache_store =
@@ -2070,6 +2471,13 @@ macro_rules! event_cache_store_integration_tests {
event_cache_store.test_find_event_relations().await;
}
#[async_test]
async fn test_find_event_relations_when_event_in_room_and_thread() {
let event_cache_store =
get_event_cache_store().await.unwrap().into_event_cache_store();
event_cache_store.test_find_event_relations_when_event_in_room_and_thread().await;
}
#[async_test]
async fn test_get_room_events() {
let event_cache_store =
@@ -2084,6 +2492,13 @@ macro_rules! event_cache_store_integration_tests {
event_cache_store.test_get_room_events_filtered().await;
}
#[async_test]
async fn test_get_room_events_with_event_in_room_and_thread() {
let event_cache_store =
get_event_cache_store().await.unwrap().into_event_cache_store();
event_cache_store.test_get_room_events_with_event_in_room_and_thread().await;
}
#[async_test]
async fn test_save_event() {
let event_cache_store =
@@ -2091,6 +2506,13 @@ macro_rules! event_cache_store_integration_tests {
event_cache_store.test_save_event().await;
}
#[async_test]
async fn test_save_event_updates_event_in_room_and_thread() {
let event_cache_store =
get_event_cache_store().await.unwrap().into_event_cache_store();
event_cache_store.test_save_event_updates_event_in_room_and_thread().await;
}
#[async_test]
async fn test_thread_vs_room_linked_chunk() {
let event_cache_store =
crates/matrix-sdk-base/src/event_cache/store/memory_store.rs
+46 -14
View File
@@ -13,7 +13,7 @@
// limitations under the License.
use std::{
collections::HashMap,
collections::{HashMap, HashSet},
sync::{Arc, RwLock as StdRwLock},
};
@@ -188,10 +188,9 @@ impl EventCacheStore for MemoryStore {
) -> Result<Option<Event>, Self::Error> {
let inner = self.inner.read().unwrap();
let event = inner
.events
.items(room_id)
.find_map(|(event, _pos)| (event.event_id()? == event_id).then_some(event.clone()));
let event = inner.events.items(room_id).find_map(|(_, (event, _pos))| {
(event.event_id()? == event_id).then_some(event.clone())
});
Ok(event)
}
@@ -204,10 +203,10 @@ impl EventCacheStore for MemoryStore {
) -> Result<Vec<(Event, Option<Position>)>, Self::Error> {
let inner = self.inner.read().unwrap();
let related_events = inner
let related_events: Vec<_> = inner
.events
.items(room_id)
.filter_map(|(event, pos)| {
.filter_map(|(linked_chunk_id, (event, pos))| {
// Must have a relation.
let (related_to, rel_type) = extract_event_relation(event.raw())?;
let rel_type = RelationType::from(rel_type.as_str());
@@ -219,14 +218,35 @@ impl EventCacheStore for MemoryStore {
// Must not be filtered out.
if let Some(filters) = &filters {
filters.contains(&rel_type).then_some((event.clone(), pos))
filters.contains(&rel_type).then_some((linked_chunk_id, (event.clone(), pos)))
} else {
Some((event.clone(), pos))
Some((linked_chunk_id, (event.clone(), pos)))
}
})
.collect();
Ok(related_events)
// Remove any duplicate events which may exist in both a room and thread
// linked chunk. Additionally, remove any position information from non-room
// linked chunks.
let mut deduplicated = HashMap::new();
for (linked_chunk_id, (event, position)) in related_events {
let event_id = event
.event_id()
.ok_or(Self::Error::InvalidData { details: String::from("missing event id") })?;
match linked_chunk_id.as_ref() {
LinkedChunkId::Room(_) => {
// Prioritize events that come from a room linked chunk
deduplicated.insert(event_id, (event, position));
}
_ => {
// Remove position information from events that come
// from any other type of linked chunk
deduplicated.entry(event_id).or_insert_with(|| (event, None));
}
}
}
Ok(deduplicated.into_values().collect())
}
async fn get_room_events(
@@ -237,17 +257,29 @@ impl EventCacheStore for MemoryStore {
) -> Result<Vec<Event>, Self::Error> {
let inner = self.inner.read().unwrap();
let event: Vec<_> = inner
let (_, event): (_, Vec<_>) = inner
.events
.items(room_id)
.map(|(event, _pos)| event.clone())
.map(|(_, (event, _pos))| event.clone())
.filter(|e| {
event_type
.is_none_or(|event_type| Some(event_type) == e.kind.event_type().as_deref())
})
.filter(|e| session_id.is_none_or(|s| Some(s) == e.kind.session_id()))
.collect();
.map(|e| {
e.event_id()
.map(|id| (id, e))
.ok_or(Self::Error::InvalidData { details: String::from("missing event id") })
})
.collect::<Result<Vec<_>>>()?
.into_iter()
.fold((HashSet::new(), Vec::new()), |(mut ids, mut es), (id, e)| {
if !ids.contains(&id) {
ids.insert(id);
es.push(e);
}
(ids, es)
});
Ok(event)
}
crates/matrix-sdk-common/CHANGELOG.md
+10
View File
@@ -7,7 +7,17 @@ All notable changes to this project will be documented in this file.
## [Unreleased] - ReleaseDate
### Features
- Add support in the `MemoryStore`'s implementation of `EventCacheStore` for
having duplicate events in a room, where each duplicate is in a different
`LinkedChunk`. This is useful, e.g., when an event is in a room and a
thread in that room.
- [**breaking**] In order to support having duplicate events in the same room (in different `LinkedChunk`'s) a few
functions were changed in `RelationalLinkedChunk`. The items in the `Iterator` returned by `RelationalLinkedChunk::items`
now also include the `LinkedChunkId` in which the `Item` was found. Additionally, `RelationalLinkedChunk::save_item`
now requires the `Item` to be `Clone` as it may be stored in multiple `LinkedChunk`s.
(#[6200](https://github.com/matrix-org/matrix-rust-sdk/pull/6200))
- [**breaking**] Added `CrossProcessLockConfig`, which can be used to configure the behavior of the cross-process lock.
`CrossProcessLock` now takes a `CrossProcessLockConfig` as an argument to its constructor instead of a `lock_holder`
value. ([#6160](https://github.com/matrix-org/matrix-rust-sdk/pull/6160))
crates/matrix-sdk-common/src/linked_chunk/relational.rs
+29 -13
View File
@@ -16,7 +16,7 @@
//! [`RelationalLinkedChunk`].
use std::{
collections::{BTreeMap, HashMap},
collections::{BTreeMap, HashMap, HashSet},
hash::Hash,
};
@@ -395,7 +395,8 @@ where
}
/// Return an iterator over all items of all linked chunks of a room, along
/// with their positions, if available.
/// with the linked chunk they are in and the position in that linked chunk,
/// if available.
///
/// The only items which will NOT have a position are those saved with
/// [`Self::save_item`].
@@ -404,26 +405,41 @@ where
pub fn items<'a>(
&'a self,
room_id: &'a RoomId,
) -> impl Iterator<Item = (&'a Item, Option<Position>)> {
) -> impl Iterator<Item = (&'a OwnedLinkedChunkId, (&'a Item, Option<Position>))> {
self.items
.iter()
.filter_map(move |(linked_chunk_id, items)| {
(linked_chunk_id.room_id() == room_id).then_some(items)
.filter(move |(linked_chunk_id, _)| linked_chunk_id.room_id() == room_id)
.flat_map(|(linked_chunk_id, items)| {
items.values().map(move |(item, pos)| (linked_chunk_id, (item, *pos)))
})
.flat_map(|items| items.values().map(|(item, pos)| (item, *pos)))
}
}
impl<ItemId, Item, Gap> RelationalLinkedChunk<ItemId, Item, Gap>
where
Item: IndexableItem<ItemId = ItemId> + Clone,
ItemId: Hash + PartialEq + Eq + Clone + Ord,
{
/// Save a single item "out-of-band" in the relational linked chunk.
pub fn save_item(&mut self, room_id: OwnedRoomId, item: Item) {
let id = item.id();
let linked_chunk_id = OwnedLinkedChunkId::Room(room_id);
let map = self.items.entry(linked_chunk_id).or_default();
if let Some(prev_value) = map.get_mut(&id) {
// If the item already exists, we keep the position.
prev_value.0 = item;
} else {
map.insert(id, (item, None));
let mut linked_chunk_ids = self
.items
.keys()
.filter(|linked_chunk_id| linked_chunk_id.room_id() == room_id)
.cloned()
.collect::<HashSet<_>>();
linked_chunk_ids.insert(OwnedLinkedChunkId::Room(room_id));
for linked_chunk_id in linked_chunk_ids {
let map = self.items.entry(linked_chunk_id).or_default();
if let Some(prev_value) = map.get_mut(&id) {
// If the item already exists, we keep the position.
prev_value.0 = item.clone();
} else {
map.insert(id.clone(), (item.clone(), None));
}
}
}
}
crates/matrix-sdk-indexeddb/CHANGELOG.md
+7
View File
@@ -8,6 +8,13 @@ All notable changes to this project will be documented in this file.
### Features
- Add support in the implementation of `EventCacheStore` for
having duplicate events in a room, where each duplicate is in a different
`LinkedChunk`. This is useful, e.g., when an event is in a room and a
thread in that room. The change involves a database migration where
the `EVENTS` object store is cleared and then modified so that the
`ROOM` index no longer requires keys to be unique.
([#6200](https://github.com/matrix-org/matrix-rust-sdk/pull/6200))
- Implement `CryptoStore::get_pending_key_bundle_details_for_room` and
`CryptoStore::get_all_rooms_pending_key_bundle`, and process
`rooms_pending_key_bundle` field in `Changes`.
crates/matrix-sdk-indexeddb/src/event_cache_store/error.rs
+4 -1
View File
@@ -32,6 +32,8 @@ pub enum IndexeddbEventCacheStoreError {
UnableToLoadChunk,
#[error("no max chunk id")]
NoMaxChunkId,
#[error("event without id")]
EventWithoutId,
#[error("transaction: {0}")]
Transaction(#[from] TransactionError),
}
@@ -68,7 +70,8 @@ impl From<IndexeddbEventCacheStoreError> for EventCacheStoreError {
| ChunksContainCycle
| ChunksContainDisjointLists
| NoMaxChunkId
| UnableToLoadChunk => Self::InvalidData { details: value.to_string() },
| UnableToLoadChunk
| EventWithoutId => Self::InvalidData { details: value.to_string() },
Transaction(inner) => inner.into(),
}
}
crates/matrix-sdk-indexeddb/src/event_cache_store/migrations.rs
+71 -4
View File
@@ -21,10 +21,10 @@ use thiserror::Error;
/// The current version and keys used in the database.
pub mod current {
use super::{Version, v2};
use super::{Version, v3};
pub const VERSION: Version = Version::V2;
pub use v2::keys;
pub const VERSION: Version = Version::V3;
pub use v3::keys;
}
/// Opens a connection to the IndexedDB database and takes care of upgrading it
@@ -56,6 +56,8 @@ pub enum Version {
V1 = 1,
/// Version 2 of the database, for details see [`v2`].
V2 = 2,
/// Version 3 of the database, for details see [`v3`].
V3 = 3,
}
impl Version {
@@ -64,7 +66,8 @@ impl Version {
match self {
Self::V0 => v0::upgrade(transaction).map(Some),
Self::V1 => v1::upgrade(transaction).map(Some),
Self::V2 => Ok(None),
Self::V2 => v2::upgrade(transaction).map(Some),
Self::V3 => Ok(None),
}
}
}
@@ -81,6 +84,7 @@ impl TryFrom<u32> for Version {
0 => Ok(Version::V0),
1 => Ok(Version::V1),
2 => Ok(Version::V2),
3 => Ok(Version::V3),
v => Err(UnknownVersionError(v)),
}
}
@@ -225,4 +229,67 @@ mod v2 {
Ok(())
}
/// Upgrade database from `v2` to `v3`
pub fn upgrade(transaction: &Transaction<'_>) -> Result<Version, Error> {
v3::update_events_object_store(transaction)?;
Ok(Version::V3)
}
}
mod v3 {
use indexed_db_futures::Build;
// Re-use all the same keys from `v2`.
pub use super::v2::keys;
use super::*;
/// Update the events object store, so that the `room` index is no longer
/// unique. This allows an event to be stored in a room twice - e.g., once
/// in the main thread and once in a side thread.
///
/// Note that this operation removes the existing events object store and
/// all of its contents.
pub fn update_events_object_store(transaction: &Transaction<'_>) -> Result<(), Error> {
remove_events_object_store(transaction)?;
create_events_object_store(transaction.db())?;
Ok(())
}
/// Remove events object store
pub fn remove_events_object_store(transaction: &Transaction<'_>) -> Result<(), Error> {
let object_store = transaction.object_store(keys::EVENTS)?;
// It is faster to clear all events first, then delete the object store rather
// than immediately deleting.
//
// For details, see https://www.artificialworlds.net/blog/2024/02/02/deleting-an-indexed-db-store-can-be-incredibly-slow-on-firefox/
object_store.clear()?;
transaction.db().delete_object_store(keys::EVENTS)?;
Ok(())
}
/// Create an object store for tracking information about events.
///
/// * Primary Key - `id`
/// * Index - `room` - tracks whether an event is in a given room
/// * Index (unique) - `position` - tracks position of an event in linked
/// chunks
/// * Index - `relation` - tracks any event to which the given event is
/// related
pub fn create_events_object_store(db: &Database) -> Result<(), Error> {
let events = db
.create_object_store(keys::EVENTS)
.with_key_path(keys::EVENTS_KEY_PATH.into())
.build()?;
let _ =
events.create_index(keys::EVENTS_ROOM, keys::EVENTS_ROOM_KEY_PATH.into()).build()?;
let _ = events
.create_index(keys::EVENTS_POSITION, keys::EVENTS_POSITION_KEY_PATH.into())
.with_unique(true)
.build()?;
let _ = events
.create_index(keys::EVENTS_RELATION, keys::EVENTS_RELATION_KEY_PATH.into())
.build()?;
Ok(())
}
}
crates/matrix-sdk-indexeddb/src/event_cache_store/mod.rs
+65 -16
View File
@@ -14,7 +14,7 @@
#![cfg_attr(not(test), allow(unused))]
use std::{rc::Rc, time::Duration};
use std::{collections::HashMap, rc::Rc, time::Duration};
use indexed_db_futures::{Build, database::Database};
#[cfg(target_family = "wasm")]
@@ -464,9 +464,9 @@ impl EventCacheStore for IndexeddbEventCacheStore {
let transaction = self.transaction(&[keys::EVENTS], IdbTransactionMode::Readonly)?;
transaction
.get_event_by_room(room_id, event_id)
.get_events_by_room(room_id, event_id)
.await
.map(|ok| ok.map(Into::into))
.map(|mut events| events.pop().map(Into::into))
.map_err(Into::into)
}
@@ -481,26 +481,60 @@ impl EventCacheStore for IndexeddbEventCacheStore {
let transaction = self.transaction(&[keys::EVENTS], IdbTransactionMode::Readonly)?;
let mut related_events = Vec::new();
let mut related_events = HashMap::<OwnedEventId, types::Event>::new();
match filters {
Some(relation_types) if !relation_types.is_empty() => {
for relation_type in relation_types {
let relation = (event_id, relation_type);
let events = transaction.get_events_by_relation(room_id, relation).await?;
for event in events {
let position = event.position().map(Into::into);
related_events.push((event.into(), position));
let Some(event_id) = event.event_id() else {
return Err(IndexeddbEventCacheStoreError::EventWithoutId);
};
match event.linked_chunk_id() {
LinkedChunkId::Room(_) => {
// Prioritize events that come from a room linked chunk
related_events.insert(event_id, event);
}
_ => {
// Remove position information from events that come
// from any other type of linked chunk
related_events
.entry(event_id)
.or_insert_with(|| event.into_out_of_band_event());
}
}
}
}
}
_ => {
for event in transaction.get_events_by_related_event(room_id, event_id).await? {
let position = event.position().map(Into::into);
related_events.push((event.into(), position));
let Some(event_id) = event.event_id() else {
return Err(IndexeddbEventCacheStoreError::EventWithoutId);
};
match event.linked_chunk_id() {
LinkedChunkId::Room(_) => {
// Prioritize events that come from a room linked chunk
related_events.insert(event_id, event);
}
_ => {
// Remove position information from events that come
// from any other type of linked chunk
related_events
.entry(event_id)
.or_insert_with(|| event.into_out_of_band_event());
}
}
}
}
}
Ok(related_events)
Ok(related_events
.into_values()
.map(|event| {
let position = event.position().map(Into::into);
(event.into(), position)
})
.collect())
}
#[instrument(skip(self))]
@@ -519,7 +553,14 @@ impl EventCacheStore for IndexeddbEventCacheStore {
transaction
.get_room_events(room_id)
.await
.map(|vec| {
.map(|mut vec| {
vec.dedup_by(|a, b| {
if let (Some(a), Some(b)) = (a.event_id(), b.event_id()) {
a == b
} else {
false
}
});
vec.into_iter()
.map(Event::from)
.filter(|e| {
@@ -546,15 +587,23 @@ impl EventCacheStore for IndexeddbEventCacheStore {
return Ok(());
};
let transaction = self.transaction(&[keys::EVENTS], IdbTransactionMode::Readwrite)?;
let event = match transaction.get_event_by_room(room_id, &event_id).await? {
Some(inner) => inner.with_content(event),
None => types::Event::OutOfBand(OutOfBandEvent {
let mut events = transaction
.get_events_by_room(room_id, &event_id)
.await?
.into_iter()
.map(|e| e.with_content(event.clone()))
.collect::<Vec<types::Event>>();
if events.is_empty() {
events.push(types::Event::OutOfBand(OutOfBandEvent {
linked_chunk_id: LinkedChunkId::Room(room_id).to_owned(),
content: event,
position: (),
}),
};
transaction.put_event(&event).await?;
}));
}
for event in events {
transaction.put_event(&event).await?;
}
transaction.commit().await?;
Ok(())
}
crates/matrix-sdk-indexeddb/src/event_cache_store/serializer/indexed_types.rs
+2 -2
View File
@@ -420,13 +420,13 @@ impl IndexedPrefixKeyBounds<Event, LinkedChunkId<'_>> for IndexedEventIdKey {
}
}
/// The value associated with the [primary key](IndexedEvent::id) of the
/// The value associated with the [`room`](IndexedEvent::room) index of the
/// [`EVENTS`][1] object store, which is constructed from:
///
/// - The (possibly) hashed Room ID
/// - The (possibly) hashed Event ID.
///
/// [1]: crate::event_cache_store::migrations::v1::create_events_object_store
/// [1]: crate::event_cache_store::migrations::v3::create_events_object_store
#[derive(Debug, Serialize, Deserialize)]
pub struct IndexedEventRoomKey(IndexedRoomId, IndexedEventId);
crates/matrix-sdk-indexeddb/src/event_cache_store/transaction.rs
+11
View File
@@ -350,6 +350,17 @@ impl<'a> IndexeddbEventCacheStoreTransaction<'a> {
self.get_item_by_key::<Event, IndexedEventRoomKey>(key).await
}
/// Query IndexedDB for events that match the given event id in the given
/// room.
pub async fn get_events_by_room(
&self,
room_id: &RoomId,
event_id: &EventId,
) -> Result<Vec<Event>, TransactionError> {
let key: IndexedEventRoomKey = self.serializer().encode_key((room_id, event_id));
self.get_items_by_key::<Event, IndexedEventRoomKey>(key).await
}
/// Query IndexedDB for events that are in the given
/// room.
pub async fn get_room_events(&self, room_id: &RoomId) -> Result<Vec<Event>, TransactionError> {
crates/matrix-sdk-indexeddb/src/event_cache_store/types.rs
+15
View File
@@ -136,6 +136,15 @@ impl Event {
}
self
}
/// Ensures that the underlying [`GenericEvent`] is an [`OutOfBandEvent`].
/// If it is not an [`OutOfBandEvent`], then it is converted into one.
pub fn into_out_of_band_event(self) -> Self {
match self {
Event::InBand(i) => Self::OutOfBand(i.into()),
o @ Event::OutOfBand(_) => o,
}
}
}
/// A generic representation of an
@@ -182,6 +191,12 @@ pub type InBandEvent = GenericEvent<Position>;
/// events which are not part of a chunk and therefore have no position.
pub type OutOfBandEvent = GenericEvent<()>;
impl From<InBandEvent> for OutOfBandEvent {
fn from(value: InBandEvent) -> Self {
Self { linked_chunk_id: value.linked_chunk_id, content: value.content, position: () }
}
}
/// A representation of [`Position`](matrix_sdk_base::linked_chunk::Position)
/// which can be stored in IndexedDB.
#[derive(Debug, Default, Copy, Clone, Serialize, Deserialize)]
crates/matrix-sdk-sqlite/src/event_cache_store.rs
+2 -81
View File
@@ -1657,13 +1657,12 @@ mod tests {
use matrix_sdk_base::{
event_cache::store::{
EventCacheStore, EventCacheStoreError, IntoEventCacheStore,
integration_tests::{EventCacheStoreIntegrationTests, make_test_event_with_event_id},
integration_tests::EventCacheStoreIntegrationTests,
},
event_cache_store_integration_tests, event_cache_store_integration_tests_time,
linked_chunk::{ChunkContent, ChunkIdentifier, LinkedChunkId, Position, Update},
linked_chunk::{ChunkIdentifier, LinkedChunkId, Update},
};
use matrix_sdk_test::{DEFAULT_TEST_ROOM_ID, async_test};
use ruma::event_id;
use tempfile::{TempDir, tempdir};
use super::SqliteEventCacheStore;
@@ -1827,84 +1826,6 @@ mod tests {
let chunks = store.load_all_chunks(linked_chunk_id).await.unwrap();
assert!(chunks.is_empty());
}
#[async_test]
async fn test_event_chunks_allows_same_event_in_room_and_thread() {
// This test verifies that the same event can appear in both a room's linked
// chunk and a thread's linked chunk. This is the real-world use case:
// a thread reply appears in both the main room timeline and the thread.
let store = get_event_cache_store().await.expect("creating cache store failed");
let room_id = *DEFAULT_TEST_ROOM_ID;
let thread_root = event_id!("$thread_root");
// Create an event that will be inserted into both the room and thread linked
// chunks.
let event = make_test_event_with_event_id(
room_id,
"thread reply",
Some(event_id!("$thread_reply")),
);
let room_linked_chunk_id = LinkedChunkId::Room(room_id);
let thread_linked_chunk_id = LinkedChunkId::Thread(room_id, thread_root);
// Insert the event into the room's linked chunk.
store
.handle_linked_chunk_updates(
room_linked_chunk_id,
vec![
Update::NewItemsChunk {
previous: None,
new: ChunkIdentifier::new(1),
next: None,
},
Update::PushItems {
at: Position::new(ChunkIdentifier::new(1), 0),
items: vec![event.clone()],
},
],
)
.await
.unwrap();
// Insert the same event into the thread's linked chunk.
store
.handle_linked_chunk_updates(
thread_linked_chunk_id,
vec![
Update::NewItemsChunk {
previous: None,
new: ChunkIdentifier::new(1),
next: None,
},
Update::PushItems {
at: Position::new(ChunkIdentifier::new(1), 0),
items: vec![event],
},
],
)
.await
.unwrap();
// Verify both entries exist by loading chunks from both linked chunk IDs.
let room_chunks = store.load_all_chunks(room_linked_chunk_id).await.unwrap();
let thread_chunks = store.load_all_chunks(thread_linked_chunk_id).await.unwrap();
assert_eq!(room_chunks.len(), 1);
assert_eq!(thread_chunks.len(), 1);
// Verify the event is in both.
assert_matches!(&room_chunks[0].content, ChunkContent::Items(events) => {
assert_eq!(events.len(), 1);
assert_eq!(events[0].event_id().as_deref(), Some(event_id!("$thread_reply")));
});
assert_matches!(&thread_chunks[0].content, ChunkContent::Items(events) => {
assert_eq!(events.len(), 1);
assert_eq!(events[0].event_id().as_deref(), Some(event_id!("$thread_reply")));
});
}
}
#[cfg(test)]