feat: properly run ingestor and block builder concurrently. also fix ctrl+c

Author: daniilrrr

Cargo.lock

@@ -33,7 +33,7 @@ async fn run(
     let sequencing_config = ingestion_config.sequencing;
     let settlement_config = ingestion_config.settlement;
 
-    let (mut sequencer_ingestor, sequencer_rx) =
+    let (mut sequencing_ingestor, sequencer_rx) =
         Ingestor::new(sequencing_config.into()).await.map_err(|e| {
             RuntimeError::Initialization(format!(
                 "Failed to create ingestor for sequencing chain: {}",
@@ -50,60 +50,53 @@ async fn run(
         })?;
 
     let slotter = Slotter::new(sequencer_rx, settlement_rx, slotting_config);
-    let slotter_rx = slotter.await.start(); // TODO(SEQ-515): refactor me to get the channel without starting the slotter already?
 
-    let block_builder = BlockBuilder::new(slotter_rx, block_builder_config).await.map_err(|e| {
+    // TODO(SEQ-515): refactor me to get the channel without starting the slotter already?
+    // TODO(SEQ-515): slotter assumes that it starts first, or else it errors here
+    let slot_rx = slotter.await.start();
+
+    let block_builder = BlockBuilder::new(slot_rx, block_builder_config).await.map_err(|e| {
         RuntimeError::Initialization(format!("Failed to create block builder: {}", e))
     })?;
 
-    let _ = block_builder.start().await;
     info!("Starting Metabased Translator");
+    let sequencing_ingestor_handle = tokio::spawn(async move {
+        if let Err(e) = sequencing_ingestor.start_polling().await {
+            error!("Ingestor error: {}", e);
+        }
+    });
 
-    sequencer_ingestor.start_polling().await.map_err(|e| {
-        error!("Sequencer ingestor error: {}", e);
-        RuntimeError::TaskFailure(e.to_string())
-    })?;
-
-    settlement_ingestor.start_polling().await.map_err(|e| {
-        error!("Settlement ingestor error: {}", e);
-        RuntimeError::TaskFailure(e.to_string())
-    })?;
+    let settlement_ingestor_handle = tokio::spawn(async move {
+        if let Err(e) = settlement_ingestor.start_polling().await {
+            error!("Ingestor error: {}", e);
+        }
+    });
 
-    // TODO(SEQ-515): Improve this
-    // // Spawn ingestor task
-    // let ingestor_handle = tokio::spawn(async move {
-    //     if let Err(e) = ingestor?.start_polling().await {
-    //         error!("Ingestor error: {}", e);
-    //     }
-    // });
-
-    // TODO(SEQ-515): Improve this
-    // // Spawn block builder task
-    // let block_builder_handle = tokio::spawn(async move {
-    //     if let Err(e) = block_builder.start().await {
-    //         error!("Block builder error: {}", e);
-    //     }
-    // });
+    // TODO(SEQ-515): Block builder doesn't error
+    let block_builder_handle = tokio::spawn(async move { block_builder.start().await });
 
     // Main control loop
     tokio::select! {
         // Wait for shutdown signal
         _ = &mut shutdown_rx => {
             info!("Metabased Translator shutting down...");
         }
-
-        // TODO(SEQ-515): Improve this
-        // // Watch for task completion/errors
-        // res = ingestor_handle => {
-        //     if let Err(e) = res {
-        //         error!("Ingestor task failed: {}", e);
-        //     }
-        // }
-        // res = block_builder_handle => {
-        //     if let Err(e) = res {
-        //         error!("Block builder task failed: {}", e);
-        //     }
-        // }
+        // Watch for task completion/errors
+        res = settlement_ingestor_handle => {
+            if let Err(e) = res {
+                error!("Settlement chain ingestor task failed: {}", e);
+            }
+        }
+        res = sequencing_ingestor_handle => {
+            if let Err(e) = res {
+                error!("Sequencing chain ingestor task failed: {}", e);
+            }
+        }
+        res = block_builder_handle => {
+            if let Err(e) = res {
+                error!("Block builder task failed: {}", e);
+            }
+        }
     }
 
     info!("Metabased Translator shutdown complete");

metabased-translator/bin/metabased-translator/src/main.rs

@@ -37,36 +37,34 @@ impl BlockBuilder {
 
     /// Start the block builder
     pub async fn start(mut self) {
-        tokio::spawn(async move {
-            while let Some(slot) = self.slotter_rx.recv().await {
-                info!("Received slot: {:?}", slot);
-
-                // Process sequencing chain blocks into mB transactions
-                let mbtxs = self.builder.parse_blocks_to_mbtxs(slot.sequencing_chain_blocks);
-
-                // TODO (SEQ-416): [OP / ARB] Process deposit transactions
-
-                // [OP / ARB] Build and submit batch
-                let batch_txn = match self.builder.build_batch_txn(mbtxs).await {
-                    Ok(txn) => txn,
-                    Err(e) => {
-                        log_error!("Error building batch transaction: {}", e);
-                        continue;
-                    }
-                };
-
-                // Submit batch transaction to mchain
-                if let Err(e) = self.mchain.submit_txn(batch_txn).await {
-                    log_error!("Error submitting transaction: {}", e);
+        while let Some(slot) = self.slotter_rx.recv().await {
+            info!("Received slot: {:?}", slot);
+
+            // Process sequencing chain blocks into mB transactions
+            let mbtxs = self.builder.parse_blocks_to_mbtxs(slot.sequencing_chain_blocks);
+
+            // TODO (SEQ-416): [OP / ARB] Process deposit transactions
+
+            // [OP / ARB] Build and submit batch
+            let batch_txn = match self.builder.build_batch_txn(mbtxs).await {
+                Ok(txn) => txn,
+                Err(e) => {
+                    log_error!("Error building batch transaction: {}", e);
                     continue;
                 }
+            };
 
-                // Mine mchain block
-                if let Err(e) = self.mchain.mine_block(slot.timestamp).await {
-                    log_error!("Error mining block: {}", e);
-                }
+            // Submit batch transaction to mchain
+            if let Err(e) = self.mchain.submit_txn(batch_txn).await {
+                log_error!("Error submitting transaction: {}", e);
+                continue;
+            }
+
+            // Mine mchain block
+            if let Err(e) = self.mchain.mine_block(slot.timestamp).await {
+                log_error!("Error mining block: {}", e);
             }
-        });
+        }
     }
 }
 

metabased-translator/crates/block-builder/src/block_builder.rs

@@ -104,7 +104,9 @@ impl MetaChainProvider {
     /// Mines a block on the `MetaChain`
     pub async fn mine_block(&self, block_timestamp_secs: u64) -> eyre::Result<()> {
         let opts = MineOptions::Options { timestamp: Some(block_timestamp_secs), blocks: Some(1) };
-        self.provider.anvil_mine_detailed(Some(opts)).await?;
+        let result = self.provider.anvil_mine_detailed(Some(opts)).await;
+        info!("{}", format!("Mined block on MetaChain {:?}", result));
+        result?;
 
         Ok(())
     }

metabased-translator/crates/block-builder/src/connectors/anvil.rs

@@ -112,7 +112,6 @@ impl Slotter {
     ///    passed)
     /// 4. Send completed slots through the returned channel
     ///
-    /// # Returns a receiver that will get slots as they are processed.
     /// TODO SEQ-480 - implement restore from DB
     pub fn start(mut self) -> Receiver<Slot> {
         self.status.store(Status::Started);
@@ -186,7 +185,6 @@ impl Slotter {
                 }
             }
         });
-
         receiver
     }
 
@@ -456,6 +454,7 @@ mod tests {
     use super::*;
     use alloy::primitives::B256;
     use std::str::FromStr;
+    use tokio::sync::mpsc::channel;
     async fn create_slotter(
         slot_start_timestamp_ms: u64,
         slot_duration_ms: u64,