Miner consensus (#878)

* chore: add consensus on the miner

* chore: temporarily send tx execution debug

* lint

src/bin/importer_offline.rs

@@ -48,7 +48,7 @@ async fn run(config: ImporterOfflineConfig) -> anyhow::Result<()> {
     // init services
     let rpc_storage = config.rpc_storage.init().await?;
     let storage = config.storage.init().await?;
-    let miner = config.miner.init(Arc::clone(&storage)).await?;
+    let miner = config.miner.init(Arc::clone(&storage), None).await?;
     let executor = config.executor.init(Arc::clone(&storage), Arc::clone(&miner), None, None).await;
 
     // init block snapshots to export

src/bin/importer_online.rs

@@ -60,7 +60,8 @@ fn main() -> anyhow::Result<()> {
 async fn run(config: ImporterOnlineConfig) -> anyhow::Result<()> {
     let storage = config.storage.init().await?;
     let relayer = config.relayer.init(Arc::clone(&storage)).await?;
-    let miner = config.miner.init(Arc::clone(&storage)).await?;
+
+    let miner = config.miner.init(Arc::clone(&storage), None).await?;
     let executor = config.executor.init(Arc::clone(&storage), Arc::clone(&miner), relayer, None).await; //XXX TODO implement the consensus here, in case of it being a follower, it should not even enter here
     let chain = Arc::new(BlockchainClient::new_http_ws(&config.base.external_rpc, config.base.external_rpc_ws.as_deref()).await?);
     let cancellation: CancellationToken = signal_handler();

src/bin/run_with_importer.rs

@@ -25,7 +25,7 @@ async fn run(config: RunWithImporterConfig) -> anyhow::Result<()> {
     let chain = Arc::new(BlockchainClient::new_http_ws(&http_url, ws_url.as_deref()).await?);
 
     let relayer = config.relayer.init(Arc::clone(&storage)).await?;
-    let miner = config.miner.init(Arc::clone(&storage)).await?;
+    let miner = config.miner.init(Arc::clone(&storage), Some(Arc::clone(&consensus))).await?;
     let executor = config.executor.init(Arc::clone(&storage), Arc::clone(&miner), relayer, Some(consensus)).await;
 
     let rpc_storage = Arc::clone(&storage);

src/config.rs

@@ -237,11 +237,11 @@ pub struct MinerConfig {
 }
 
 impl MinerConfig {
-    pub async fn init(&self, storage: Arc<StratusStorage>) -> anyhow::Result<Arc<BlockMiner>> {
+    pub async fn init(&self, storage: Arc<StratusStorage>, consensus: Option<Arc<Consensus>>) -> anyhow::Result<Arc<BlockMiner>> {
         tracing::info!(config = ?self, "starting block miner");
 
         // create miner
-        let miner = BlockMiner::new(Arc::clone(&storage), self.block_time);
+        let miner = BlockMiner::new(Arc::clone(&storage), self.block_time, consensus);
         let miner = Arc::new(miner);
 
         // enable genesis block

src/eth/block_miner.rs

@@ -10,6 +10,7 @@ use nonempty::NonEmpty;
 use tokio::runtime::Handle;
 use tokio::sync::broadcast;
 
+use super::Consensus;
 use crate::eth::primitives::Block;
 use crate::eth::primitives::BlockHeader;
 use crate::eth::primitives::BlockNumber;
@@ -39,18 +40,22 @@ pub struct BlockMiner {
 
     /// Broadcasts transaction logs events.
     pub notifier_logs: broadcast::Sender<LogMined>,
+
+    /// Consensus logic.
+    consensus: Option<Arc<Consensus>>,
 }
 
 impl BlockMiner {
     /// Creates a new [`BlockMiner`].
-    pub fn new(storage: Arc<StratusStorage>, block_time: Option<Duration>) -> Self {
+    pub fn new(storage: Arc<StratusStorage>, block_time: Option<Duration>, consensus: Option<Arc<Consensus>>) -> Self {
         tracing::info!("starting block miner");
         Self {
             storage,
             block_time,
             notifier_pending_txs: broadcast::channel(u16::MAX as usize).0,
             notifier_blocks: broadcast::channel(u16::MAX as usize).0,
             notifier_logs: broadcast::channel(u16::MAX as usize).0,
+            consensus,
         }
     }
 
@@ -109,7 +114,11 @@ impl BlockMiner {
     pub async fn save_execution(&self, tx_execution: TransactionExecution) -> anyhow::Result<()> {
         let tx_hash = tx_execution.hash();
 
-        self.storage.save_execution(tx_execution).await?;
+        self.storage.save_execution(tx_execution.clone()).await?;
+        if let Some(consensus) = &self.consensus {
+            let execution = format!("{:?}", tx_execution.clone());
+            consensus.sender.send(execution).await.unwrap();
+        }
         let _ = self.notifier_pending_txs.send(tx_hash);
 
         Ok(())

src/main.rs

@@ -14,7 +14,7 @@ async fn run(config: StratusConfig) -> anyhow::Result<()> {
     // init services
     let storage = config.storage.init().await?;
     let relayer = config.relayer.init(Arc::clone(&storage)).await?;
-    let miner = config.miner.init(Arc::clone(&storage)).await?;
+    let miner = config.miner.init(Arc::clone(&storage), None).await?;
     let executor = config.executor.init(Arc::clone(&storage), Arc::clone(&miner), relayer, None).await;
     let cancellation = signal_handler();
 

tests/test_import_external_snapshot_common.rs

@@ -155,7 +155,7 @@ pub async fn execute_test(
     // init services
     let storage = Arc::new(StratusStorage::new(Arc::new(InMemoryTemporaryStorage::new()), Arc::new(perm_storage)));
     let relayer = config.relayer.init(Arc::clone(&storage)).await.unwrap();
-    let miner = config.miner.init(Arc::clone(&storage)).await.unwrap();
+    let miner = config.miner.init(Arc::clone(&storage), None).await.unwrap();
     let executor = config.executor.init(Arc::clone(&storage), Arc::clone(&miner), relayer, None).await;
 
     // execute and mine