Simplification of the dual sequencer/no sequencer logic

Author: fulmicoton-dd

quickwit/quickwit-indexing/src/actors/metrics_pipeline/mod.rs

@@ -50,3 +50,14 @@ pub use parquet_uploader::ParquetUploader;
 pub use pipeline::{MetricsPipeline, MetricsPipelineParams};
 pub use processed_parquet_batch::ProcessedParquetBatch;
 pub(crate) use publisher_impl::METRICS_PUBLISHER_NAME;
+
+#[cfg(test)]
+/// Spawn a `Sequencer<Publisher>` in front of the given publisher mailbox.
+pub(crate) fn spawn_sequencer_for_test(
+    universe: &quickwit_actors::Universe,
+    publisher_mailbox: quickwit_actors::Mailbox<crate::actors::Publisher>,
+) -> quickwit_actors::Mailbox<crate::actors::Sequencer<crate::actors::Publisher>> {
+    let sequencer = crate::actors::Sequencer::new(publisher_mailbox);
+    let (sequencer_mailbox, _sequencer_handle) = universe.spawn_builder().spawn(sequencer);
+    sequencer_mailbox
+}

quickwit/quickwit-indexing/src/actors/metrics_pipeline/parquet_doc_processor.rs

@@ -524,7 +524,7 @@ mod tests {
         use quickwit_storage::RamStorage;
 
         use crate::actors::metrics_pipeline::{ParquetIndexer, ParquetPackager, ParquetUploader};
-        use crate::actors::{Publisher, SplitsUpdateMailbox, UploaderType};
+        use crate::actors::{Publisher, UploaderType};
 
         let universe = Universe::with_accelerated_time();
         let temp_dir = tempfile::tempdir().unwrap();
@@ -533,11 +533,12 @@ mod tests {
         let mock_metastore = MockMetastoreService::new();
         let ram_storage = StdArc::new(RamStorage::default());
         let (publisher_mailbox, _publisher_inbox) = universe.create_test_mailbox::<Publisher>();
+        let sequencer_mailbox = super::super::spawn_sequencer_for_test(&universe, publisher_mailbox);
         let uploader = ParquetUploader::new(
             UploaderType::IndexUploader,
             quickwit_proto::metastore::MetastoreServiceClient::from_mock(mock_metastore),
             ram_storage,
-            SplitsUpdateMailbox::Publisher(publisher_mailbox),
+            sequencer_mailbox,
             4,
         );
         let (uploader_mailbox, _uploader_handle) = universe.spawn_builder().spawn(uploader);

quickwit/quickwit-indexing/src/actors/metrics_pipeline/parquet_e2e_test.rs

@@ -37,8 +37,7 @@ use quickwit_proto::types::IndexUid;
 use quickwit_storage::RamStorage;
 
 use crate::actors::{
-    ParquetDocProcessor, ParquetIndexer, ParquetPackager, ParquetUploader, Publisher,
-    SplitsUpdateMailbox, UploaderType,
+    ParquetDocProcessor, ParquetIndexer, ParquetPackager, ParquetUploader, Publisher, UploaderType,
 };
 use crate::models::RawDocBatch;
 
@@ -162,11 +161,13 @@ async fn test_metrics_pipeline_e2e() {
     );
     let (publisher_mailbox, publisher_handle) = universe.spawn_builder().spawn(publisher);
 
+    let sequencer_mailbox = super::spawn_sequencer_for_test(&universe, publisher_mailbox);
+
     let uploader = ParquetUploader::new(
         UploaderType::IndexUploader,
         metastore_client.clone(),
         ram_storage,
-        SplitsUpdateMailbox::Publisher(publisher_mailbox),
+        sequencer_mailbox,
         4,
     );
     let (uploader_mailbox, _uploader_handle) = universe.spawn_builder().spawn(uploader);

quickwit/quickwit-indexing/src/actors/metrics_pipeline/parquet_indexer.rs

@@ -554,7 +554,7 @@ mod tests {
 
     use super::*;
     use crate::actors::metrics_pipeline::{ParquetPackager, ParquetUploader};
-    use crate::actors::{Publisher, SplitsUpdateMailbox, UploaderType};
+    use crate::actors::{Publisher, UploaderType};
 
     /// Create a test ParquetUploader and return its mailbox.
     fn create_test_uploader(
@@ -563,12 +563,13 @@ mod tests {
         let mock_metastore = MockMetastoreService::new();
         let ram_storage = Arc::new(RamStorage::default());
         let (publisher_mailbox, _publisher_inbox) = universe.create_test_mailbox::<Publisher>();
+        let sequencer_mailbox = super::super::spawn_sequencer_for_test(universe, publisher_mailbox);
 
         let uploader = ParquetUploader::new(
             UploaderType::IndexUploader,
             quickwit_proto::metastore::MetastoreServiceClient::from_mock(mock_metastore),
             ram_storage,
-            SplitsUpdateMailbox::Publisher(publisher_mailbox),
+            sequencer_mailbox,
             4,
         );
         universe.spawn_builder().spawn(uploader)
@@ -587,12 +588,13 @@ mod tests {
 
         let ram_storage = Arc::new(RamStorage::default());
         let (publisher_mailbox, _publisher_inbox) = universe.create_test_mailbox::<Publisher>();
+        let sequencer_mailbox = super::super::spawn_sequencer_for_test(universe, publisher_mailbox);
 
         let uploader = ParquetUploader::new(
             UploaderType::IndexUploader,
             quickwit_proto::metastore::MetastoreServiceClient::from_mock(mock_metastore),
             ram_storage,
-            SplitsUpdateMailbox::Publisher(publisher_mailbox),
+            sequencer_mailbox,
             4,
         );
         universe.spawn_builder().spawn(uploader)
@@ -915,11 +917,12 @@ mod tests {
 
         let ram_storage = Arc::new(RamStorage::default());
         let (publisher_mailbox, _publisher_inbox) = universe.create_test_mailbox::<Publisher>();
+        let sequencer_mailbox = super::super::spawn_sequencer_for_test(&universe, publisher_mailbox);
         let uploader = ParquetUploader::new(
             UploaderType::IndexUploader,
             quickwit_proto::metastore::MetastoreServiceClient::from_mock(mock_metastore),
             ram_storage,
-            SplitsUpdateMailbox::Publisher(publisher_mailbox),
+            sequencer_mailbox,
             4,
         );
         let (uploader_mailbox, uploader_handle) = universe.spawn_builder().spawn(uploader);

quickwit/quickwit-indexing/src/actors/metrics_pipeline/parquet_packager.rs

@@ -244,7 +244,7 @@ mod tests {
     use quickwit_storage::RamStorage;
 
     use super::*;
-    use crate::actors::{Publisher, SplitsUpdateMailbox, UploaderType};
+    use crate::actors::{Publisher, UploaderType};
 
     fn create_test_uploader(
         universe: &Universe,
@@ -256,12 +256,13 @@ mod tests {
 
         let ram_storage = Arc::new(RamStorage::default());
         let (publisher_mailbox, _publisher_inbox) = universe.create_test_mailbox::<Publisher>();
+        let sequencer_mailbox = super::super::spawn_sequencer_for_test(universe, publisher_mailbox);
 
         let uploader = ParquetUploader::new(
             UploaderType::IndexUploader,
             quickwit_proto::metastore::MetastoreServiceClient::from_mock(mock_metastore),
             ram_storage,
-            SplitsUpdateMailbox::Publisher(publisher_mailbox),
+            sequencer_mailbox,
             4,
         );
         universe.spawn_builder().spawn(uploader)