From fa5f4d5c8a52c7b994c40f2964d9e010f04b1922 Mon Sep 17 00:00:00 2001 From: Azur Date: Tue, 7 Jul 2026 17:46:24 +0200 Subject: [PATCH 1/2] fix(search): wait for merge threads before dropping the bulk writer Signed-off-by: Azur --- crates/matrix-sdk-search/changelog.d/6731.fixed.md | 1 + crates/matrix-sdk-search/src/index/mod.rs | 1 + crates/matrix-sdk-search/src/writer.rs | 4 ++++ 3 files changed, 6 insertions(+) create mode 100644 crates/matrix-sdk-search/changelog.d/6731.fixed.md diff --git a/crates/matrix-sdk-search/changelog.d/6731.fixed.md b/crates/matrix-sdk-search/changelog.d/6731.fixed.md new file mode 100644 index 00000000000..c23fa996b45 --- /dev/null +++ b/crates/matrix-sdk-search/changelog.d/6731.fixed.md @@ -0,0 +1 @@ +`bulk_execute` now waits for the index writers merge threads before dropping it, avoiding spurious `couldn't find segment in SegmentManager` warnings and index fragmentation diff --git a/crates/matrix-sdk-search/src/index/mod.rs b/crates/matrix-sdk-search/src/index/mod.rs index 7b6e2200010..9bd2328ec33 100644 --- a/crates/matrix-sdk-search/src/index/mod.rs +++ b/crates/matrix-sdk-search/src/index/mod.rs @@ -317,6 +317,7 @@ impl RoomIndex { } self.commit_and_reload(&mut writer)?; + writer.wait_merging_threads()?; Ok(()) } diff --git a/crates/matrix-sdk-search/src/writer.rs b/crates/matrix-sdk-search/src/writer.rs index 1bee9044813..c4989006483 100644 --- a/crates/matrix-sdk-search/src/writer.rs +++ b/crates/matrix-sdk-search/src/writer.rs @@ -46,4 +46,8 @@ impl SearchIndexWriter { self.last_commit_opstamp = self.inner.commit()?; // TODO: This is blocking. Handle it. Ok(self.last_commit_opstamp) } + + pub(crate) fn wait_merging_threads(self) -> Result<(), TantivyError> { + self.inner.wait_merging_threads() + } } From 9cd1bbf0e42726892d04c649d111312b5bc0c9b4 Mon Sep 17 00:00:00 2001 From: Azur Date: Fri, 10 Jul 2026 18:35:31 +0200 Subject: [PATCH 2/2] fixup! fix(search): wait for merge threads before dropping the bulk writer Signed-off-by: Azur --- .../changelog.d/6731.fixed.md | 2 +- crates/matrix-sdk-search/src/index/mod.rs | 34 +++++++++++++++++++ crates/matrix-sdk-search/src/writer.rs | 10 +++--- 3 files changed, 40 insertions(+), 6 deletions(-) diff --git a/crates/matrix-sdk-search/changelog.d/6731.fixed.md b/crates/matrix-sdk-search/changelog.d/6731.fixed.md index c23fa996b45..c161c6999de 100644 --- a/crates/matrix-sdk-search/changelog.d/6731.fixed.md +++ b/crates/matrix-sdk-search/changelog.d/6731.fixed.md @@ -1 +1 @@ -`bulk_execute` now waits for the index writers merge threads before dropping it, avoiding spurious `couldn't find segment in SegmentManager` warnings and index fragmentation +`RoomIndex::bulk_execute` now waits for the index writers merge threads before dropping it, avoiding spurious "couldn't find segment in SegmentManager" warnings and index fragmentation diff --git a/crates/matrix-sdk-search/src/index/mod.rs b/crates/matrix-sdk-search/src/index/mod.rs index 9bd2328ec33..75fe4dfe9b8 100644 --- a/crates/matrix-sdk-search/src/index/mod.rs +++ b/crates/matrix-sdk-search/src/index/mod.rs @@ -451,6 +451,40 @@ mod tests { Ok(()) } + #[test] + fn test_bulk_execute_indexes_all_events() -> Result<(), Box> { + let room_id = room_id!("!room_id:localhost"); + let mut index = RoomIndexBuilder::new_in_memory(room_id).build(); + + let event_id_1 = event_id!("$event_id_1:localhost"); + let event_id_2 = event_id!("$event_id_2:localhost"); + let user_id = user_id!("@user_id:localhost"); + let f = EventFactory::new().room(room_id).sender(user_id); + + let ops = vec![ + RoomIndexOperation::Add( + f.text_msg("This is a sentence") + .event_id(event_id_1) + .into_original_sync_room_message_event(), + ), + RoomIndexOperation::Add( + f.text_msg("Another sentence") + .event_id(event_id_2) + .into_original_sync_room_message_event(), + ), + ]; + + index.bulk_execute(ops)?; + + let result: HashSet<_> = + index.search("sentence", 10, None)?.into_iter().map(|(_score, id)| id).collect(); + let expected: HashSet<_> = + [event_id_1.to_owned(), event_id_2.to_owned()].into_iter().collect(); + assert_eq!(result, expected, "bulk_execute did not index all events: {result:?}"); + + Ok(()) + } + #[test] fn test_search_empty_index() -> Result<(), Box> { let room_id = room_id!("!room_id:localhost"); diff --git a/crates/matrix-sdk-search/src/writer.rs b/crates/matrix-sdk-search/src/writer.rs index c4989006483..dd5677a0e90 100644 --- a/crates/matrix-sdk-search/src/writer.rs +++ b/crates/matrix-sdk-search/src/writer.rs @@ -28,26 +28,26 @@ pub(crate) struct SearchIndexWriter { } impl SearchIndexWriter { - pub(crate) fn new(writer: IndexWriter, schema: RoomMessageSchema) -> Self { + pub fn new(writer: IndexWriter, schema: RoomMessageSchema) -> Self { Self { last_commit_opstamp: writer.commit_opstamp(), inner: writer, schema } } - pub(crate) fn add(&self, document: TantivyDocument) -> Result { + pub fn add(&self, document: TantivyDocument) -> Result { Ok(self.inner.add_document(document)?) // TODO: This is blocking. Handle // it. } - pub(crate) fn remove(&self, event_id: &EventId) { + pub fn remove(&self, event_id: &EventId) { self.inner .delete_term(Term::from_field_text(self.schema.deletion_key(), event_id.as_str())); } - pub(crate) fn commit(&mut self) -> Result { + pub fn commit(&mut self) -> Result { self.last_commit_opstamp = self.inner.commit()?; // TODO: This is blocking. Handle it. Ok(self.last_commit_opstamp) } - pub(crate) fn wait_merging_threads(self) -> Result<(), TantivyError> { + pub fn wait_merging_threads(self) -> Result<(), TantivyError> { self.inner.wait_merging_threads() } }