.

rs/p2p/consensus_manager/src/lib.rs

@@ -25,20 +25,15 @@ mod metrics;
 mod receiver;
 mod sender;
 
-type StartConsensusManagerFn =
-    Box<dyn FnOnce(Arc<dyn Transport>, watch::Receiver<SubnetTopology>) -> Vec<Shutdown>>;
+type StartConsensusManagerFn = Box<dyn FnOnce(Arc<dyn Transport>) -> Vec<Shutdown>>;
 
 pub struct AbortableBroadcastChannelManager(Vec<StartConsensusManagerFn>);
 
 impl AbortableBroadcastChannelManager {
-    pub fn start(
-        self,
-        transport: Arc<dyn Transport>,
-        topology_watcher: watch::Receiver<SubnetTopology>,
-    ) -> Vec<Shutdown> {
+    pub fn start(self, transport: Arc<dyn Transport>) -> Vec<Shutdown> {
         let mut ret = vec![];
         for client in self.0 {
-            ret.append(&mut client(transport.clone(), topology_watcher.clone()));
+            ret.append(&mut client(transport.clone()));
         }
         ret
     }
@@ -77,6 +72,7 @@ impl AbortableBroadcastChannelBuilder {
         D: ArtifactAssembler<Artifact, WireArtifact>,
     >(
         &mut self,
+        topology_watcher: watch::Receiver<SubnetTopology>,
         (assembler, assembler_router): (F, Router),
         slot_limit: usize,
     ) -> (
@@ -102,7 +98,7 @@ impl AbortableBroadcastChannelBuilder {
         let rt_handle = self.rt_handle.clone();
         let metrics_registry = self.metrics_registry.clone();
 
-        let builder = move |transport: Arc<dyn Transport>, topology_watcher| {
+        let builder = move |transport: Arc<dyn Transport>| {
             start_consensus_manager(
                 log,
                 &metrics_registry,

rs/p2p/consensus_manager/tests/test.rs

@@ -342,21 +342,25 @@ fn load_test(
     let mut jhs = vec![];
     let mut nodes = vec![];
     let mut cms = vec![];
+    let (nodes, topology_watcher) = fully_connected_localhost_subnet(rt.handle(), log, id, nodes);
     for i in 0..num_peers {
         let node = node_test_id(i);
         let processor = TestConsensus::new(log.clone(), node, 256 * (i as usize + 1), i % 2 == 0);
-        let (jh, cm) =
-            start_consensus_manager(no_op_logger(), rt.handle().clone(), processor.clone());
+        let (jh, cm) = start_consensus_manager(
+            no_op_logger(),
+            rt.handle().clone(),
+            processor.clone(),
+            topology_watcher.clone(),
+        );
         jhs.push(jh);
         let (r, m) = cm.build();
         nodes.push((node, r));
         cms.push((node, m));
         node_advert_map.insert(node, processor);
     }
-    let (nodes, topology_watcher) = fully_connected_localhost_subnet(rt.handle(), log, id, nodes);
     for ((node1, transport), (node2, cm)) in nodes.into_iter().zip(cms.into_iter()) {
         assert!(node1 == node2);
-        cm.start(transport, topology_watcher.clone());
+        cm.start(transport);
     }
 
     rt.block_on(async move {

rs/p2p/test_utils/src/lib.rs

@@ -447,6 +447,7 @@ pub fn start_consensus_manager(
     log: ReplicaLogger,
     rt_handle: Handle,
     processor: TestConsensus<U64Artifact>,
+    topology_watcher: tokio::sync::watch::Receiver<SubnetTopology>,
 ) -> (
     Box<dyn JoinGuard>,
     ic_consensus_manager::AbortableBroadcastChannelBuilder,
@@ -467,7 +468,8 @@ pub fn start_consensus_manager(
         rt_handle.clone(),
         MetricsRegistry::default(),
     );
-    let (outbound_tx, inbound_rx) = cm1.abortable_broadcast_channel(downloader, usize::MAX);
+    let (outbound_tx, inbound_rx) =
+        cm1.abortable_broadcast_channel(topology_watcher, downloader, usize::MAX);
 
     let artifact_processor_jh = start_test_processor(
         outbound_tx,

rs/p2p/test_utils/src/turmoil.rs

@@ -381,8 +381,11 @@ pub fn add_transport_to_sim<F>(
                     bouncer_factory,
                     MetricsRegistry::default(),
                 );
-                let (outbound_tx, inbound_tx) =
-                    consensus_builder.abortable_broadcast_channel(downloader, usize::MAX);
+                let (outbound_tx, inbound_tx) = consensus_builder.abortable_broadcast_channel(
+                    topology_watcher_clone.clone(),
+                    downloader,
+                    usize::MAX,
+                );
 
                 let artifact_processor_jh = start_test_processor(
                     outbound_tx,
@@ -412,7 +415,7 @@ pub fn add_transport_to_sim<F>(
             ));
 
             if let Some((_, con_manager)) = con {
-                con_manager.start(transport.clone(), topology_watcher_clone.clone());
+                con_manager.start(transport.clone());
             }
 
             if let Some(state_sync_manager) = state_sync_manager {