cleanup(consensus): Remove Broadcaster and wrapper around NetworkManager (#6794)

# Description of change

This PR simplifies the implementation of managing network setup in the
Starfish consensus implementation. It removes unnecessary
`NetworkManager` trait, because there is currently one supported network
type - Tonic. Since Tonic supports streaming, Broadcaster has also been
removed as it was only used for network types that don't support
streaming.

## Links to any relevant issues

Fixes #6423 

## Type of change

Choose a type of change, and delete any options that are not relevant.

- Enhancement (a non-breaking change which adds functionality)

## How the change has been tested

Unable to test because starfish implementation is not yet in a working
condition.

Make sure to provide instructions for the maintainer as well as any
relevant configurations.

- [x] Basic tests (linting, compilation, formatting, unit/integration
tests)
- [ ] Patch-specific tests (correctness, functionality coverage)

Author: piotrm50

crates/iota-core/src/consensus_manager/starfish_manager.rs

@@ -8,7 +8,6 @@ use async_trait::async_trait;
 use fastcrypto::ed25519;
 use iota_config::NodeConfig;
 use iota_metrics::{RegistryID, RegistryService, monitored_mpsc::unbounded_channel};
-use iota_protocol_config::ConsensusNetwork;
 use iota_types::{
     committee::EpochId,
     iota_system_state::epoch_start_iota_system_state::EpochStartSystemStateTrait,
@@ -83,21 +82,6 @@ impl StarfishManager {
         store_path.push(format!("{}", epoch));
         store_path
     }
-
-    fn pick_network(&self, epoch_store: &AuthorityPerEpochStore) -> ConsensusNetwork {
-        if let Ok(type_str) = std::env::var("CONSENSUS_NETWORK") {
-            match type_str.to_lowercase().as_str() {
-                "tonic" => return ConsensusNetwork::Tonic,
-                _ => {
-                    info!(
-                        "Invalid consensus network type {} in env var. Continue to use the value from protocol config.",
-                        type_str
-                    );
-                }
-            }
-        }
-        epoch_store.protocol_config().consensus_network()
-    }
 }
 
 #[async_trait]
@@ -115,7 +99,6 @@ impl ConsensusManagerTrait for StarfishManager {
         let committee: Committee = system_state.get_starfish_committee();
         let epoch = epoch_store.epoch();
         let protocol_config = epoch_store.protocol_config();
-        let network_type = self.pick_network(&epoch_store);
 
         let Some(_guard) = RunningLockGuard::acquire_start(
             &self.metrics,
@@ -183,7 +166,6 @@ impl ConsensusManagerTrait for StarfishManager {
         }
 
         let authority = ConsensusAuthority::start(
-            network_type,
             own_index,
             committee.clone(),
             parameters.clone(),

crates/iota-core/src/unit_tests/starfish_manager_tests.rs

@@ -10,7 +10,7 @@ use iota_types::messages_checkpoint::{
     CertifiedCheckpointSummary, CheckpointContents, CheckpointSummary,
 };
 use prometheus::Registry;
-use tokio::{sync::mpsc, time::sleep};
+use tokio::{sync::mpsc, task::yield_now, time::sleep};
 
 use crate::{
     authority::{AuthorityState, test_authority_builder::TestAuthorityBuilder},
@@ -132,5 +132,8 @@ async fn test_starfish_manager() {
 
         // THEN
         assert!(!manager.is_running().await);
+
+        // Yield to ensure that RocksDB releases the lock before the next iteration.
+        yield_now().await;
     }
 }