fix(core): reduce block messages with many connections

base_layer/core/src/base_node/service/service.rs

@@ -649,7 +649,7 @@ async fn handle_outbound_block(
     exclude_peers: Vec<NodeId>,
 ) -> Result<(), CommsInterfaceError> {
     let result = outbound_message_service
-        .flood(
+        .propagate(
             NodeDestination::Unknown,
             OutboundEncryption::ClearText,
             exclude_peers,

comms/dht/examples/memory_net/utilities.rs

@@ -332,6 +332,7 @@ pub async fn do_network_wide_propagation(nodes: &mut [TestNode], origin_node_ind
                                 OutboundEncryption::ClearText,
                                 vec![msg.source_peer.node_id.clone()],
                                 OutboundDomainMessage::new(&0i32, public_msg),
+                                "Memory net example".to_string(),
                             )
                             .await
                             .unwrap();

comms/dht/examples/propagation_stress.rs

@@ -129,6 +129,7 @@ async fn prompt(node: &CommsNode, dht: &Dht) -> anyhow::Result<()> {
                         // Don't send directly to peer
                         vec![opts.peer.node_id.clone()],
                         msg,
+                        "Example stress".to_string(),
                     )
                     .await?
             },

comms/dht/src/config.rs

@@ -178,7 +178,7 @@ impl Default for DhtConfig {
             num_neighbouring_nodes: 8,
             num_random_nodes: 4,
             minimize_connections: false,
-            propagation_factor: 4,
+            propagation_factor: 20,
             broadcast_factor: 8,
             outbound_buffer_size: 20,
             saf: Default::default(),

comms/dht/src/outbound/requester.rs

@@ -136,12 +136,14 @@ impl OutboundMessageRequester {
         encryption: OutboundEncryption,
         exclude_peers: Vec<NodeId>,
         message: OutboundDomainMessage<T>,
+        source_info: String,
     ) -> Result<MessageSendStates, DhtOutboundError>
     where
         T: prost::Message,
     {
         self.send_message(
             SendMessageParams::new()
+                .with_debug_info(source_info)
                 .propagate(destination.clone(), exclude_peers)
                 .with_encryption(encryption)
                 .with_destination(destination)

comms/dht/tests/dht.rs

@@ -366,6 +366,7 @@ async fn test_dht_propagate_dedup() {
             OutboundEncryption::encrypt_for(node_D.node_identity().public_key().clone()),
             vec![],
             out_msg,
+            String::new(),
         )
         .await
         .unwrap();
@@ -655,6 +656,7 @@ async fn test_dht_repropagate() {
             OutboundEncryption::ClearText,
             vec![],
             out_msg.clone(),
+            String::new(),
         )
         .await
         .unwrap();