fix(core): reduce block messages with many connections (#6602)

Description
---
fix(core): reduce block messages with many connections

Motivation and Context
---
When a seed node is connected to many hundreds of base node peers, it
propagates blocks on all connections.
This PR changes this to send to 20 random peers. Default
propagation_factor is changed from 4 to 20.

How Has This Been Tested?
---

What process can a PR reviewer use to test or verify this change?
---

<!-- Checklist -->
<!-- 1. Is the title of your PR in the form that would make nice release
notes? The title, excluding the conventional commit
tag, will be included exactly as is in the CHANGELOG, so please think
about it carefully. -->


Breaking Changes
---

- [x] None
- [ ] Requires data directory on base node to be deleted
- [ ] Requires hard fork
- [ ] Other - Please specify

<!-- Does this include a breaking change? If so, include this line as a
footer -->
<!-- BREAKING CHANGE: Description what the user should do, e.g. delete a
database, resync the chain -->

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();