Text compaction race test (#37262)

Copies over structure of vector search compaction tests to text search.

GitOrigin-RevId: 684672faacf1dcd8f4fd1f9845fb0ff2fe215ced

Author: emmaling27

crates/database/src/index_workers/search_compactor.rs

@@ -108,11 +108,12 @@ impl<RT: Runtime, T: SearchIndex> SearchIndexCompactor<RT, T> {
         let mut to_build = vec![];
         let mut tx = self.database.begin(Identity::system()).await?;
 
-        // Skip compaction on empty tables.
-        for index_doc in IndexModel::new(&mut tx)
+        let non_empty_search_indexes = IndexModel::new(&mut tx)
             .get_all_non_empty_search_indexes()
-            .await?
-        {
+            .await?;
+
+        // Skip compaction on empty tables.
+        for index_doc in non_empty_search_indexes {
             let (index_id, index_metadata) = index_doc.into_id_and_value();
             let Some(config) = T::get_config(index_metadata.config) else {
                 continue;
@@ -147,11 +148,19 @@ impl<RT: Runtime, T: SearchIndex> SearchIndexCompactor<RT, T> {
                     &config.developer_config,
                     &self.config,
                 )?,
-                _ => continue,
+                _ => {
+                    tracing::info!(
+                        "Skipping {:?} index for compaction: {name:?} because it is not \
+                         backfilling",
+                        Self::search_type()
+                    );
+                    continue;
+                },
             };
             if let Some((mut segments_to_compact, compaction_reason)) = maybe_segments_to_compact {
                 tracing::info!(
-                    "Queueing {:?} index for compaction: {name:?}",
+                    "Queueing {:?} index for compaction: {name:?} for reason: \
+                     {compaction_reason:?}",
                     Self::search_type()
                 );
                 // Choose segments to compact at random.

crates/database/src/tests/text_test_utils.rs

@@ -17,6 +17,7 @@ use common::{
         ParsedDocument,
         ResolvedDocument,
     },
+    pause::PauseController,
     persistence::PersistenceReader,
     query::{
         Query,
@@ -35,6 +36,7 @@ use common::{
     },
     version::MIN_NPM_VERSION_FOR_FUZZY_SEARCH,
 };
+use futures::try_join;
 use maplit::btreeset;
 use must_let::must_let;
 use search::{
@@ -79,6 +81,7 @@ use crate::{
     Transaction,
 };
 
+#[derive(Clone)]
 pub struct TextFixtures {
     pub rt: TestRuntime,
     pub storage: Arc<dyn Storage>,
@@ -345,6 +348,29 @@ impl TextFixtures {
         }
         Ok(values)
     }
+
+    pub async fn run_compaction_during_flush(
+        &self,
+        pause: PauseController,
+        label: &'static str,
+    ) -> anyhow::Result<()> {
+        let mut flusher = self.new_search_flusher();
+        let hold_guard = pause.hold(label);
+        let flush = flusher.step();
+        let compactor = self.new_compactor();
+        let compact_during_flush = async move {
+            if let Some(pause_guard) = hold_guard.wait_for_blocked().await {
+                let (metrics, _) = compactor.step().await?;
+                for (_index_name, num_segments_compacted) in metrics {
+                    assert!(num_segments_compacted > 0);
+                }
+                pause_guard.unpause();
+            };
+            Ok::<(), anyhow::Error>(())
+        };
+        try_join!(flush, compact_during_flush)?;
+        Ok(())
+    }
 }
 
 const TABLE_NAME: &str = "table";

crates/database/src/text_index_worker/flusher.rs

@@ -192,6 +192,7 @@ mod tests {
             IndexConfig,
             IndexMetadata,
         },
+        pause::PauseController,
         runtime::testing::TestRuntime,
         types::{
             IndexName,
@@ -207,6 +208,10 @@ mod tests {
     };
 
     use crate::{
+        index_workers::{
+            search_compactor::CompactionConfig,
+            search_flusher::FLUSH_RUNNING_LABEL,
+        },
         tests::text_test_utils::{
             add_document,
             IndexData,
@@ -991,4 +996,56 @@ mod tests {
 
         Ok(())
     }
+
+    #[convex_macro::test_runtime]
+    async fn concurrent_compaction_and_flush_new_segment_propagates_deletes(
+        rt: TestRuntime,
+        pause: PauseController,
+    ) -> anyhow::Result<()> {
+        let config = CompactionConfig::default();
+        let min_compaction_segments = config.min_compaction_segments;
+        let config = CompactionConfig {
+            // Treat everything as a large segment
+            small_segment_threshold_bytes: 0,
+            ..config
+        };
+        let fixtures = TextFixtures::new_with_config(rt.clone(), config).await?;
+        let index_data = fixtures.enabled_text_index().await?;
+
+        let IndexData { index_name, .. } = index_data;
+
+        // Create enough segments to trigger compaction.
+        let mut deleted_doc_ids = vec![];
+        for _ in 0..min_compaction_segments {
+            deleted_doc_ids.push(fixtures.add_document("test").await?);
+            fixtures.backfill().await?;
+        }
+
+        // Queue up deletes for all existing segments, and one new document that will
+        // cause the flusher to write a new segment.
+        for doc_id in &deleted_doc_ids {
+            fixtures.replace_document(*doc_id, "updated").await?;
+        }
+        let _non_deleted_id = fixtures.add_document("test").await?;
+
+        // Run the compactor / flusher concurrently in a way where the compactor
+        // wins the race.
+        fixtures
+            .run_compaction_during_flush(pause, FLUSH_RUNNING_LABEL)
+            .await?;
+
+        // Verify we propagate the new deletes to the compacted segment and retain our
+        // new segment.
+        let segments = fixtures.get_segments_metadata(index_name).await?;
+        assert_eq!(2, segments.len());
+
+        let (compacted_segment, new_segment): (Vec<_>, Vec<_>) = segments
+            .into_iter()
+            .partition(|segment| segment.num_deleted_documents > 0);
+        assert_eq!(compacted_segment.len(), 1);
+        assert_eq!(new_segment.len(), 1);
+        // TODO Verify segment contents
+
+        Ok(())
+    }
 }

crates/database/src/text_index_worker/text_meta.rs

@@ -106,6 +106,7 @@ impl SegmentType<TextSearchIndex> for FragmentedTextSegment {
         Ok(self.size_bytes_total)
     }
 }
+
 #[derive(Clone)]
 pub struct BuildTextIndexArgs {
     pub search_storage: Arc<dyn Storage>,