fix(search): wait for merge threads before dropping the bulk writer

crates/matrix-sdk-search/changelog.d/6731.fixed.md

@@ -0,0 +1 @@
+`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

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(())
     }
@@ -450,6 +451,40 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_bulk_execute_indexes_all_events() -> Result<(), Box<dyn Error>> {
+        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<dyn Error>> {
         let room_id = room_id!("!room_id:localhost");

crates/matrix-sdk-search/src/writer.rs

@@ -28,22 +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<OpStamp, IndexError> {
+    pub fn add(&self, document: TantivyDocument) -> Result<OpStamp, IndexError> {
         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<OpStamp, TantivyError> {
+    pub fn commit(&mut self) -> Result<OpStamp, TantivyError> {
         self.last_commit_opstamp = self.inner.commit()?; // TODO: This is blocking. Handle it.
         Ok(self.last_commit_opstamp)
     }
+
+    pub fn wait_merging_threads(self) -> Result<(), TantivyError> {
+        self.inner.wait_merging_threads()
+    }
 }