Reduce Validation and Proposal message relaying

Builds/CMake/RippledCore.cmake

@@ -867,6 +867,7 @@ target_sources (rippled PRIVATE
   src/test/overlay/cluster_test.cpp
   src/test/overlay/short_read_test.cpp
   src/test/overlay/compression_test.cpp
+  src/test/overlay/reduce_relay_test.cpp
   #[===============================[
      test sources:
        subdir: peerfinder

bin/ci/ubuntu/build-and-test.sh

@@ -136,6 +136,7 @@ else
     # ORDER matters here...sorted in approximately
     # descending execution time (longest running tests at top)
     declare -a manual_tests=(
+        'ripple.ripple_data.reduce_relay_simulate'
         'ripple.ripple_data.digest'
         'ripple.tx.Offer_manual'
         'ripple.app.PayStrandAllPairs'

src/ripple/app/consensus/RCLConsensus.cpp

@@ -157,7 +157,7 @@ RCLConsensus::Adaptor::share(RCLCxPeerPos const& peerPos)
     auto const sig = peerPos.signature();
     prop.set_signature(sig.data(), sig.size());
 
-    app_.overlay().relay(prop, peerPos.suppressionID());
+    app_.overlay().relay(prop, peerPos.suppressionID(), peerPos.publicKey());
 }
 
 void

src/ripple/app/misc/HashRouter.cpp

@@ -49,12 +49,18 @@ HashRouter::addSuppression(uint256 const& key)
 
 bool
 HashRouter::addSuppressionPeer(uint256 const& key, PeerShortID peer)
+{
+    return addSuppressionPeerWithStatus(key, peer).first;
+}
+
+std::pair<bool, std::optional<Stopwatch::time_point>>
+HashRouter::addSuppressionPeerWithStatus(const uint256& key, PeerShortID peer)
 {
     std::lock_guard lock(mutex_);
 
     auto result = emplace(key);
     result.first.addPeer(peer);
-    return result.second;
+    return {result.second, result.first.relayed()};
 }
 
 bool
@@ -110,14 +116,14 @@ HashRouter::setFlags(uint256 const& key, int flags)
 
 auto
 HashRouter::shouldRelay(uint256 const& key)
-    -> boost::optional<std::set<PeerShortID>>
+    -> std::optional<std::set<PeerShortID>>
 {
     std::lock_guard lock(mutex_);
 
     auto& s = emplace(key).first;
 
     if (!s.shouldRelay(suppressionMap_.clock().now(), holdTime_))
-        return boost::none;
+        return {};
 
     return s.releasePeerSet();
 }

src/ripple/app/misc/HashRouter.h

@@ -97,6 +97,13 @@ class HashRouter
             return std::move(peers_);
         }
 
+        /** Return seated relay time point if the message has been relayed */
+        std::optional<Stopwatch::time_point>
+        relayed() const
+        {
+            return relayed_;
+        }
+
         /** Determines if this item should be relayed.
 
             Checks whether the item has been recently relayed.
@@ -142,8 +149,8 @@ class HashRouter
         std::set<PeerShortID> peers_;
         // This could be generalized to a map, if more
         // than one flag needs to expire independently.
-        boost::optional<Stopwatch::time_point> relayed_;
-        boost::optional<Stopwatch::time_point> processed_;
+        std::optional<Stopwatch::time_point> relayed_;
+        std::optional<Stopwatch::time_point> processed_;
         std::uint32_t recoveries_ = 0;
     };
 
@@ -185,6 +192,14 @@ class HashRouter
     bool
     addSuppressionPeer(uint256 const& key, PeerShortID peer);
 
+    /** Add a suppression peer and get message's relay status.
+     * Return pair:
+     * element 1: true if the peer is added.
+     * element 2: optional is seated to the relay time point or
+     * is unseated if has not relayed yet. */
+    std::pair<bool, std::optional<Stopwatch::time_point>>
+    addSuppressionPeerWithStatus(uint256 const& key, PeerShortID peer);
+
     bool
     addSuppressionPeer(uint256 const& key, PeerShortID peer, int& flags);
 
@@ -214,11 +229,11 @@ class HashRouter
             return `true` again until the hold time has expired.
             The internal set of peers will also be reset.
 
-        @return A `boost::optional` set of peers which do not need to be
+        @return A `std::optional` set of peers which do not need to be
             relayed to. If the result is uninitialized, the item should
             _not_ be relayed.
     */
-    boost::optional<std::set<PeerShortID>>
+    std::optional<std::set<PeerShortID>>
     shouldRelay(uint256 const& key);
 
     /** Determines whether the hashed item should be recovered

src/ripple/core/Config.h

@@ -179,6 +179,12 @@ class Config : public BasicConfig
     // Thread pool configuration
     std::size_t WORKERS = 0;
 
+    // Reduce-relay - these parameters are experimental.
+    // Enable reduce-relay functionality
+    bool REDUCE_RELAY_ENABLE = false;
+    // Send squelch message to peers
+    bool REDUCE_RELAY_SQUELCH = false;
+
     // These override the command line client settings
     boost::optional<beast::IP::Endpoint> rpc_ip;
 

src/ripple/core/ConfigSections.h

@@ -70,6 +70,7 @@ struct ConfigSection
 #define SECTION_PATH_SEARCH_MAX "path_search_max"
 #define SECTION_PEER_PRIVATE "peer_private"
 #define SECTION_PEERS_MAX "peers_max"
+#define SECTION_REDUCE_RELAY "reduce_relay"
 #define SECTION_RELAY_PROPOSALS "relay_proposals"
 #define SECTION_RELAY_VALIDATIONS "relay_validations"
 #define SECTION_RPC_STARTUP "rpc_startup"

src/ripple/core/impl/Config.cpp

@@ -481,6 +481,13 @@ Config::loadFromString(std::string const& fileContents)
     if (getSingleSection(secConfig, SECTION_COMPRESSION, strTemp, j_))
         COMPRESSION = beast::lexicalCastThrow<bool>(strTemp);
 
+    if (exists(SECTION_REDUCE_RELAY))
+    {
+        auto sec = section(SECTION_REDUCE_RELAY);
+        REDUCE_RELAY_ENABLE = sec.value_or("enable", false);
+        REDUCE_RELAY_SQUELCH = sec.value_or("squelch", false);
+    }
+
     if (getSingleSection(
             secConfig, SECTION_AMENDMENT_MAJORITY_TIME, strTemp, j_))
     {

src/ripple/overlay/Message.h

@@ -21,6 +21,7 @@
 #define RIPPLE_OVERLAY_MESSAGE_H_INCLUDED
 
 #include <ripple/overlay/Compression.h>
+#include <ripple/protocol/PublicKey.h>
 #include <ripple/protocol/messages.h>
 #include <boost/asio/buffer.hpp>
 #include <boost/asio/buffers_iterator.hpp>
@@ -55,8 +56,13 @@ class Message : public std::enable_shared_from_this<Message>
     /** Constructor
      * @param message Protocol message to serialize
      * @param type Protocol message type
+     * @param validator Public Key of the source validator for Validation or
+     * Proposal message. Used to check if the message should be squelched.
      */
-    Message(::google::protobuf::Message const& message, int type);
+    Message(
+        ::google::protobuf::Message const& message,
+        int type,
+        boost::optional<PublicKey> const& validator = {});
 
     /** Retrieve the packed message data. If compressed message is requested but
      * the message is not compressible then the uncompressed buffer is returned.
@@ -74,11 +80,19 @@ class Message : public std::enable_shared_from_this<Message>
         return category_;
     }
 
+    /** Get the validator's key */
+    boost::optional<PublicKey> const&
+    getValidatorKey() const
+    {
+        return validatorKey_;
+    }
+
 private:
     std::vector<uint8_t> buffer_;
     std::vector<uint8_t> bufferCompressed_;
     std::size_t category_;
     std::once_flag once_flag_;
+    boost::optional<PublicKey> validatorKey_;
 
     /** Set the payload header
      * @param in Pointer to the payload

src/ripple/overlay/Overlay.h

@@ -133,7 +133,7 @@ class Overlay : public Stoppable, public beast::PropertyStream::Source
 
     /** Returns the peer with the matching short id, or null. */
     virtual std::shared_ptr<Peer>
-    findPeerByShortID(Peer::id_t const& id) = 0;
+    findPeerByShortID(Peer::id_t const& id) const = 0;
 
     /** Returns the peer with the matching public key, or null. */
     virtual std::shared_ptr<Peer>
@@ -147,13 +147,29 @@ class Overlay : public Stoppable, public beast::PropertyStream::Source
     virtual void
     broadcast(protocol::TMValidation& m) = 0;
 
-    /** Relay a proposal. */
-    virtual void
-    relay(protocol::TMProposeSet& m, uint256 const& uid) = 0;
-
-    /** Relay a validation. */
-    virtual void
-    relay(protocol::TMValidation& m, uint256 const& uid) = 0;
+    /** Relay a proposal.
+     * @param m the serialized proposal
+     * @param uid the id used to identify this proposal
+     * @param validator The pubkey of the validator that issued this proposal
+     * @return the set of peers which have already sent us this proposal
+     */
+    virtual std::set<Peer::id_t>
+    relay(
+        protocol::TMProposeSet& m,
+        uint256 const& uid,
+        PublicKey const& validator) = 0;
+
+    /** Relay a validation.
+     * @param m the serialized validation
+     * @param uid the id used to identify this validation
+     * @param validator The pubkey of the validator that issued this validation
+     * @return the set of peers which have already sent us this validation
+     */
+    virtual std::set<Peer::id_t>
+    relay(
+        protocol::TMValidation& m,
+        uint256 const& uid,
+        PublicKey const& validator) = 0;
 
     /** Visit every active peer.
      *