fix some flaky tests

quickwit/quickwit-indexing/src/actors/indexing_service.rs

@@ -23,7 +23,7 @@ use futures::TryStreamExt;
 use itertools::Itertools;
 use quickwit_actors::{
     Actor, ActorContext, ActorExitStatus, ActorHandle, ActorState, Handler, Healthz, Mailbox,
-    Observation,
+    Observation, SendError,
 };
 use quickwit_cluster::Cluster;
 use quickwit_common::fs::get_cache_directory_path;
@@ -539,12 +539,17 @@ impl IndexingService {
                 // The queue capacity of the merge pipeline is unbounded, so `.send_message(...)`
                 // should not block.
                 // We avoid using `.quit()` here because it waits for the actor to exit.
-                merge_pipeline_handle
+                // In some case the pipeline could already be shutting down, in which case we can
+                // receive a Disconnected
+                match merge_pipeline_handle
                     .handle
                     .mailbox()
                     .send_message(FinishPendingMergesAndShutdownPipeline)
                     .await
-                    .expect("merge pipeline mailbox should not be full");
+                {
+                    Ok(_) | Err(SendError::Disconnected) => (),
+                    Err(SendError::Full) => panic!("merge pipeline mailbox should not be full"),
+                }
             }
         }
         // Finally, we remove the completed or failed merge pipelines.

quickwit/quickwit-indexing/src/actors/merge_executor.rs

@@ -796,7 +796,6 @@ mod tests {
         delete_task_executor_handle
             .process_pending_and_observe()
             .await;
-        universe.sleep(*quickwit_actors::HEARTBEAT).await;
 
         let packager_msgs: Vec<IndexedSplitBatch> = merge_packager_inbox.drain_for_test_typed();
         if !result_docs.is_empty() {
@@ -853,7 +852,6 @@ mod tests {
                 |split| split.split_state == quickwit_metastore::SplitState::MarkedForDeletion
             ));
         }
-        universe.sleep(*quickwit_actors::HEARTBEAT).await;
         test_sandbox.assert_quit().await;
         universe.assert_quit().await;
         Ok(())

quickwit/quickwit-integration-tests/src/tests/update_tests/doc_mapping_tests.rs

@@ -101,7 +101,7 @@ async fn validate_search_across_doc_mapping_updates(
         )
         .await
         .unwrap();
-
+    tokio::time::sleep(Duration::from_millis(50)).await;
     sandbox
         .local_ingest(index_id, ingest_after_update)
         .await