feat(consensus): run network manager in consensus (#1809)

crates/consensus_manager/Cargo.toml

@@ -21,5 +21,6 @@ papyrus_protobuf.workspace = true
 serde.workspace = true
 starknet_batcher_types.workspace = true
 starknet_sequencer_infra.workspace = true
+tokio.workspace = true
 tracing.workspace = true
 validator.workspace = true

crates/consensus_manager/src/consensus_manager.rs

@@ -8,7 +8,7 @@ use futures::SinkExt;
 use libp2p::PeerId;
 use papyrus_consensus::types::{BroadcastConsensusMessageChannel, ConsensusError};
 use papyrus_consensus_orchestrator::sequencer_consensus_context::SequencerConsensusContext;
-use papyrus_network::network_manager::BroadcastTopicClient;
+use papyrus_network::network_manager::{BroadcastTopicClient, NetworkManager};
 use papyrus_network_types::network_types::BroadcastedMessageMetadata;
 use papyrus_protobuf::consensus::ConsensusMessage;
 use starknet_batcher_types::communication::SharedBatcherClient;
@@ -18,6 +18,10 @@ use tracing::{error, info};
 
 use crate::config::ConsensusManagerConfig;
 
+// TODO(Dan, Guy): move to config.
+pub const BROADCAST_BUFFER_SIZE: usize = 100;
+pub const NETWORK_TOPIC: &str = "consensus_proposals";
+
 #[derive(Clone)]
 pub struct ConsensusManager {
     pub config: ConsensusManagerConfig,
@@ -30,21 +34,33 @@ impl ConsensusManager {
     }
 
     pub async fn run(&self) -> Result<(), ConsensusError> {
+        let network_manager =
+            NetworkManager::new(self.config.consensus_config.network_config.clone(), None);
         let context = SequencerConsensusContext::new(
             Arc::clone(&self.batcher_client),
             self.config.consensus_config.num_validators,
         );
 
-        papyrus_consensus::run_consensus(
+        let mut network_handle = tokio::task::spawn(network_manager.run());
+        let consensus_task = papyrus_consensus::run_consensus(
             context,
             self.config.consensus_config.start_height,
             self.config.consensus_config.validator_id,
             self.config.consensus_config.consensus_delay,
             self.config.consensus_config.timeouts.clone(),
             create_fake_network_channels(),
             futures::stream::pending(),
-        )
-        .await
+        );
+
+        tokio::select! {
+            consensus_result = consensus_task => {
+                match consensus_result {
+                    Ok(_) => panic!("Consensus task finished unexpectedly"),
+                    Err(e) => Err(e),
+                }
+            },
+            _ = &mut network_handle => panic!("Consensus Network handle finished unexpectedly"),
+        }
     }
 }
 

crates/papyrus_node/src/bin/run_consensus.rs

@@ -4,6 +4,7 @@
 //! Expects to receive 2 groupings of arguments:
 //! 1. TestConfig - these are prefixed with `--test.` in the command.
 //! 2. NodeConfig - any argument lacking the above prefix is assumed to be in NodeConfig.
+
 use clap::Parser;
 use futures::stream::StreamExt;
 use papyrus_consensus::config::ConsensusConfig;

crates/papyrus_node/src/run.rs

@@ -2,7 +2,7 @@
 #[path = "run_test.rs"]
 mod run_test;
 
-use std::future::pending;
+use std::future;
 use std::process::exit;
 use std::sync::Arc;
 use std::time::Duration;
@@ -152,7 +152,7 @@ async fn spawn_rpc_server(
     _pending_classes: Arc<RwLock<PendingClasses>>,
     _storage_reader: StorageReader,
 ) -> anyhow::Result<JoinHandle<anyhow::Result<()>>> {
-    Ok(tokio::spawn(pending()))
+    Ok(tokio::spawn(future::pending()))
 }
 
 fn spawn_monitoring_server(
@@ -178,7 +178,7 @@ fn spawn_consensus(
 ) -> anyhow::Result<JoinHandle<anyhow::Result<()>>> {
     let (Some(config), Some(network_manager)) = (config, network_manager) else {
         info!("Consensus is disabled.");
-        return Ok(tokio::spawn(pending()));
+        return Ok(tokio::spawn(future::pending()));
     };
     let config = config.clone();
     debug!("Consensus configuration: {config:?}");
@@ -248,7 +248,7 @@ async fn spawn_sync_client(
         (Some(_), Some(_)) => {
             panic!("One of --sync.#is_none or --p2p_sync.#is_none must be turned on");
         }
-        (None, None) => tokio::spawn(pending()),
+        (None, None) => tokio::spawn(future::pending()),
         (Some(sync_config), None) => {
             let configs = (sync_config, config.central.clone(), config.base_layer.clone());
             let storage = (storage_reader.clone(), storage_writer);
@@ -294,7 +294,7 @@ fn spawn_p2p_sync_server(
 ) -> JoinHandle<anyhow::Result<()>> {
     let Some(network_manager) = network_manager else {
         info!("P2P Sync is disabled.");
-        return tokio::spawn(pending());
+        return tokio::spawn(future::pending());
     };
 
     let header_server_receiver = network_manager
@@ -404,7 +404,7 @@ async fn run_threads(
     } else {
         match resources.maybe_network_manager {
             Some(manager) => tokio::spawn(async move { Ok(manager.run().await?) }),
-            None => tokio::spawn(pending()),
+            None => tokio::spawn(future::pending()),
         }
     };
     tokio::select! {
@@ -460,7 +460,7 @@ fn spawn_storage_metrics_collector(
     interval: Duration,
 ) -> JoinHandle<anyhow::Result<()>> {
     if !collect_metrics {
-        return tokio::spawn(pending());
+        return tokio::spawn(future::pending());
     }
 
     tokio::spawn(

crates/sequencing/papyrus_consensus/src/manager.rs

@@ -30,6 +30,7 @@ use crate::types::{
 // TODO(dvir): add test for this.
 #[instrument(skip_all, level = "info")]
 #[allow(missing_docs)]
+#[allow(clippy::too_many_arguments)]
 pub async fn run_consensus<ContextT, SyncReceiverT>(
     mut context: ContextT,
     start_height: BlockNumber,