extend proptests, fix race condition (#1107)

* extend proptests, fix race condition
* cargo fmt

src/core/segment_id.rs

@@ -108,6 +108,12 @@ impl fmt::Debug for SegmentId {
     }
 }
 
+impl fmt::Display for SegmentId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "Seg({:?})", self.short_uuid_string())
+    }
+}
+
 impl PartialOrd for SegmentId {
     fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
         Some(self.cmp(other))

src/indexer/index_writer.rs

@@ -803,6 +803,7 @@ mod tests {
     use crate::schema::TextFieldIndexing;
     use crate::schema::TextOptions;
     use crate::schema::STORED;
+    use crate::schema::TEXT;
     use crate::schema::{self, IndexRecordOption, FAST, INDEXED, STRING};
     use crate::DocAddress;
     use crate::Index;
@@ -810,6 +811,15 @@ mod tests {
     use crate::Term;
     use crate::{IndexSettings, IndexSortByField, Order};
 
+    const LOREM: &str = "Doc Lorem ipsum dolor sit amet, consectetur adipiscing elit, sed \
+             do eiusmod tempor incididunt ut labore et dolore magna aliqua. \
+             Ut enim ad minim veniam, quis nostrud exercitation ullamco \
+             laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure \
+             dolor in reprehenderit in voluptate velit esse cillum dolore eu \
+             fugiat nulla pariatur. Excepteur sint occaecat cupidatat non \
+             proident, sunt in culpa qui officia deserunt mollit anim id est \
+             laborum.";
+
     #[test]
     fn test_operations_group() {
         // an operations group with 2 items should cause 3 opstamps 0, 1, and 2.
@@ -1396,6 +1406,9 @@ mod tests {
                 )
                 .set_stored(),
         );
+
+        let large_text_field = schema_builder.add_text_field("large_text_field", TEXT | STORED);
+
         let multi_numbers = schema_builder.add_u64_field(
             "multi_numbers",
             IntOptions::default()
@@ -1427,7 +1440,7 @@ mod tests {
                 IndexingOp::AddDoc { id } => {
                     let facet = Facet::from(&("/cola/".to_string() + &id.to_string()));
                     index_writer
-                        .add_document(doc!(id_field=>id, multi_numbers=> id, multi_numbers => id, text_field => id.to_string(), facet_field => facet));
+                        .add_document(doc!(id_field=>id, multi_numbers=> id, multi_numbers => id, text_field => id.to_string(), facet_field => facet, large_text_field=> LOREM));
                 }
                 IndexingOp::DeleteDoc { id } => {
                     index_writer.delete_term(Term::from_field_u64(id_field, id));
@@ -1438,13 +1451,16 @@ mod tests {
             }
         }
         index_writer.commit()?;
+
         let searcher = index.reader()?.searcher();
         if force_merge {
+            index_writer.wait_merging_threads()?;
+            let mut index_writer = index.writer_for_tests()?;
             let segment_ids = index
                 .searchable_segment_ids()
                 .expect("Searchable segments failed.");
             if segment_ids.len() >= 2 {
-                assert!(block_on(index_writer.merge(&segment_ids)).is_ok());
+                block_on(index_writer.merge(&segment_ids)).unwrap();
                 assert!(index_writer.wait_merging_threads().is_ok());
             }
         }
@@ -1521,7 +1537,7 @@ mod tests {
                 .parse_query(term)
                 .unwrap();
             let top_docs: Vec<(f32, DocAddress)> =
-                searcher.search(&query, &TopDocs::with_limit(3)).unwrap();
+                searcher.search(&query, &TopDocs::with_limit(1000)).unwrap();
 
             top_docs.iter().map(|el| el.1).collect::<Vec<_>>()
         };

src/indexer/segment_manager.rs

@@ -32,6 +32,12 @@ impl SegmentRegisters {
         } else if self.committed.contains_all(segment_ids) {
             Some(SegmentsStatus::Committed)
         } else {
+            warn!(
+                "segment_ids: {:?}, committed_ids: {:?}, uncommitted_ids {:?}",
+                segment_ids,
+                self.committed.segment_ids(),
+                self.uncommitted.segment_ids()
+            );
             None
         }
     }
@@ -58,21 +64,6 @@ impl Debug for SegmentManager {
     }
 }
 
-pub fn get_mergeable_segments(
-    in_merge_segment_ids: &HashSet<SegmentId>,
-    segment_manager: &SegmentManager,
-) -> (Vec<SegmentMeta>, Vec<SegmentMeta>) {
-    let registers_lock = segment_manager.read();
-    (
-        registers_lock
-            .committed
-            .get_mergeable_segments(in_merge_segment_ids),
-        registers_lock
-            .uncommitted
-            .get_mergeable_segments(in_merge_segment_ids),
-    )
-}
-
 impl SegmentManager {
     pub fn from_segments(
         segment_metas: Vec<SegmentMeta>,
@@ -86,6 +77,20 @@ impl SegmentManager {
         }
     }
 
+    pub fn get_mergeable_segments(
+        &self,
+        in_merge_segment_ids: &HashSet<SegmentId>,
+    ) -> (Vec<SegmentMeta>, Vec<SegmentMeta>) {
+        let registers_lock = self.read();
+        (
+            registers_lock
+                .committed
+                .get_mergeable_segments(in_merge_segment_ids),
+            registers_lock
+                .uncommitted
+                .get_mergeable_segments(in_merge_segment_ids),
+        )
+    }
     /// Returns all of the segment entries (committed or uncommitted)
     pub fn segment_entries(&self) -> Vec<SegmentEntry> {
         let registers_lock = self.read();

src/indexer/segment_register.rs

@@ -4,6 +4,7 @@ use crate::indexer::delete_queue::DeleteCursor;
 use crate::indexer::segment_entry::SegmentEntry;
 use std::collections::HashMap;
 use std::collections::HashSet;
+use std::fmt::Display;
 use std::fmt::{self, Debug, Formatter};
 
 /// The segment register keeps track
@@ -29,6 +30,16 @@ impl Debug for SegmentRegister {
         Ok(())
     }
 }
+impl Display for SegmentRegister {
+    fn fmt(&self, f: &mut Formatter<'_>) -> Result<(), fmt::Error> {
+        write!(f, "SegmentRegister(")?;
+        for k in self.segment_states.keys() {
+            write!(f, "{}, ", k.short_uuid_string())?;
+        }
+        write!(f, ")")?;
+        Ok(())
+    }
+}
 
 impl SegmentRegister {
     pub fn clear(&mut self) {
@@ -46,6 +57,10 @@ impl SegmentRegister {
             .collect()
     }
 
+    pub fn segment_ids(&self) -> Vec<SegmentId> {
+        self.segment_states.keys().cloned().collect()
+    }
+
     pub fn segment_entries(&self) -> Vec<SegmentEntry> {
         self.segment_states.values().cloned().collect()
     }

src/indexer/segment_updater.rs

@@ -1,4 +1,4 @@
-use super::segment_manager::{get_mergeable_segments, SegmentManager};
+use super::segment_manager::SegmentManager;
 use crate::core::Index;
 use crate::core::IndexMeta;
 use crate::core::IndexSettings;
@@ -527,10 +527,14 @@ impl SegmentUpdater {
         }))
     }
 
-    async fn consider_merge_options(&self) {
+    pub(crate) fn get_mergeable_segments(&self) -> (Vec<SegmentMeta>, Vec<SegmentMeta>) {
         let merge_segment_ids: HashSet<SegmentId> = self.merge_operations.segment_in_merge();
-        let (committed_segments, uncommitted_segments) =
-            get_mergeable_segments(&merge_segment_ids, &self.segment_manager);
+        self.segment_manager
+            .get_mergeable_segments(&merge_segment_ids)
+    }
+
+    async fn consider_merge_options(&self) {
+        let (committed_segments, uncommitted_segments) = self.get_mergeable_segments();
 
         // Committed segments cannot be merged with uncommitted_segments.
         // We therefore consider merges using these two sets of segments independently.