From b7c8810ac6a5bc41888fc653ae4d2d1e2dd6d743 Mon Sep 17 00:00:00 2001
From: Stan Bondi <sdbondi@users.noreply.github.com>
Date: Fri, 4 Oct 2024 12:12:16 +0200
Subject: [PATCH] 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 | 2 +-
 comms/dht/examples/memory_net/utilities.rs       | 1 +
 comms/dht/examples/propagation_stress.rs         | 1 +
 comms/dht/src/config.rs                          | 2 +-
 comms/dht/src/outbound/requester.rs              | 2 ++
 comms/dht/tests/dht.rs                           | 2 ++
 6 files changed, 8 insertions(+), 2 deletions(-)

diff --git a/base_layer/core/src/base_node/service/service.rs b/base_layer/core/src/base_node/service/service.rs
index 5a98eee9ba..8d270a4165 100644
--- a/base_layer/core/src/base_node/service/service.rs
+++ b/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,
diff --git a/comms/dht/examples/memory_net/utilities.rs b/comms/dht/examples/memory_net/utilities.rs
index 7812a53e8a..b140bf5592 100644
--- a/comms/dht/examples/memory_net/utilities.rs
+++ b/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();
diff --git a/comms/dht/examples/propagation_stress.rs b/comms/dht/examples/propagation_stress.rs
index 865a5b45d1..7b8c6df2eb 100644
--- a/comms/dht/examples/propagation_stress.rs
+++ b/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?
             },
diff --git a/comms/dht/src/config.rs b/comms/dht/src/config.rs
index 069c77b9f1..ad8eaebf18 100644
--- a/comms/dht/src/config.rs
+++ b/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(),
diff --git a/comms/dht/src/outbound/requester.rs b/comms/dht/src/outbound/requester.rs
index 5ce56537c0..02826b71fa 100644
--- a/comms/dht/src/outbound/requester.rs
+++ b/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)
diff --git a/comms/dht/tests/dht.rs b/comms/dht/tests/dht.rs
index 09ed64aa60..71aa20f759 100644
--- a/comms/dht/tests/dht.rs
+++ b/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();