From fc8f1a4ca7078aa37069bfbc08dabed9b2c0482e Mon Sep 17 00:00:00 2001
From: Age Manning <Age@AgeManning.com>
Date: Thu, 7 Mar 2024 20:48:51 +1100
Subject: [PATCH] Attempt to publish to at least mesh_n peers (#5357)

* Code improvements

* Fix gossipsub tests

* Merge latest unstable

* Differentiate errors and better scoring

* Attempt to publish to mesh_n peers
---
 .../src/gossipsub/behaviour.rs                | 37 ++++++++++++++++---
 .../src/gossipsub/behaviour/tests.rs          |  4 +-
 .../lighthouse_network/src/gossipsub/error.rs |  3 ++
 3 files changed, 37 insertions(+), 7 deletions(-)

diff --git a/beacon_node/lighthouse_network/src/gossipsub/behaviour.rs b/beacon_node/lighthouse_network/src/gossipsub/behaviour.rs
index 9769adca2..455c285c8 100644
--- a/beacon_node/lighthouse_network/src/gossipsub/behaviour.rs
+++ b/beacon_node/lighthouse_network/src/gossipsub/behaviour.rs
@@ -635,9 +635,33 @@ where
                     || !self.score_below_threshold(p, |ts| ts.publish_threshold).0
             }));
         } else {
-            match self.mesh.get(&raw_message.topic) {
+            match self.mesh.get(&topic_hash) {
                 // Mesh peers
                 Some(mesh_peers) => {
+                    // We have a mesh set. We want to make sure to publish to at least `mesh_n`
+                    // peers (if possible).
+                    let needed_extra_peers = self.config.mesh_n().saturating_sub(mesh_peers.len());
+
+                    if needed_extra_peers > 0 {
+                        // We don't have `mesh_n` peers in our mesh, we will randomly select extras
+                        // and publish to them.
+
+                        // Get a random set of peers that are appropriate to send messages too.
+                        let peer_list = get_random_peers(
+                            &self.connected_peers,
+                            &topic_hash,
+                            needed_extra_peers,
+                            |peer| {
+                                !mesh_peers.contains(peer)
+                                    && !self.explicit_peers.contains(peer)
+                                    && !self
+                                        .score_below_threshold(peer, |pst| pst.publish_threshold)
+                                        .0
+                            },
+                        );
+                        recipient_peers.extend(peer_list);
+                    }
+
                     recipient_peers.extend(mesh_peers);
                 }
                 // Gossipsub peers
@@ -729,10 +753,14 @@ where
             }
         }
 
-        if publish_failed {
+        if recipient_peers.is_empty() {
             return Err(PublishError::InsufficientPeers);
         }
 
+        if publish_failed {
+            return Err(PublishError::AllQueuesFull(recipient_peers.len()));
+        }
+
         tracing::debug!(message=%msg_id, "Published message");
 
         if let Some(metrics) = self.metrics.as_mut() {
@@ -2203,10 +2231,9 @@ where
                         if outbound <= self.config.mesh_outbound_min() {
                             // do not remove anymore outbound peers
                             continue;
-                        } else {
-                            // an outbound peer gets removed
-                            outbound -= 1;
                         }
+                        // an outbound peer gets removed
+                        outbound -= 1;
                     }
 
                     // remove the peer
diff --git a/beacon_node/lighthouse_network/src/gossipsub/behaviour/tests.rs b/beacon_node/lighthouse_network/src/gossipsub/behaviour/tests.rs
index eb006e529..f191d38f5 100644
--- a/beacon_node/lighthouse_network/src/gossipsub/behaviour/tests.rs
+++ b/beacon_node/lighthouse_network/src/gossipsub/behaviour/tests.rs
@@ -741,8 +741,8 @@ fn test_publish_without_flood_publishing() {
     let config: Config = Config::default();
     assert_eq!(
         publishes.len(),
-        config.mesh_n_low(),
-        "Should send a publish message to all known peers"
+        config.mesh_n(),
+        "Should send a publish message to at least mesh_n peers"
     );
 
     assert!(
diff --git a/beacon_node/lighthouse_network/src/gossipsub/error.rs b/beacon_node/lighthouse_network/src/gossipsub/error.rs
index d00e1ec6d..df3332bc9 100644
--- a/beacon_node/lighthouse_network/src/gossipsub/error.rs
+++ b/beacon_node/lighthouse_network/src/gossipsub/error.rs
@@ -36,6 +36,9 @@ pub enum PublishError {
     MessageTooLarge,
     /// The compression algorithm failed.
     TransformFailed(std::io::Error),
+    /// Messages could not be sent because all queues for peers were full. The usize represents the
+    /// number of peers that have full queues.
+    AllQueuesFull(usize),
 }
 
 impl std::fmt::Display for PublishError {