Merge pull request #1613 from AntelopeIO/hs-1605-msg-infra

IF: Infrastructure for hs message filtering

libraries/chain/include/eosio/chain/hotstuff.hpp

@@ -57,6 +57,13 @@ namespace eosio::chain {
 
    using hs_message = std::variant<hs_vote_message, hs_proposal_message, hs_new_block_message, hs_new_view_message>;
 
+   enum class hs_message_warning {
+      discarded,               // default code for dropped messages (irrelevant, redundant, ...)
+      duplicate_signature,     // same message signature already seen
+      invalid_signature,       // invalid message signature
+      invalid                  // invalid message (other reason)
+   };
+
    struct finalizer_state {
       bool chained_mode = false;
       fc::sha256 b_leaf;

libraries/hotstuff/chain_pacemaker.cpp

@@ -118,12 +118,18 @@ namespace eosio { namespace hotstuff {
       _head_block_state = chain->head_block_state();
    }
 
-   void chain_pacemaker::register_bcast_function(std::function<void(const chain::hs_message&)> broadcast_hs_message) {
+   void chain_pacemaker::register_bcast_function(std::function<void(const std::optional<uint32_t>&, const chain::hs_message&)> broadcast_hs_message) {
       FC_ASSERT(broadcast_hs_message, "on_hs_message must be provided");
-      std::lock_guard g( _hotstuff_global_mutex ); // not actually needed but doesn't hurt
+      // no need to std::lock_guard g( _hotstuff_global_mutex ); here since pre-comm init
       bcast_hs_message = std::move(broadcast_hs_message);
    }
 
+   void chain_pacemaker::register_warn_function(std::function<void(const uint32_t, const chain::hs_message_warning&)> warning_hs_message) {
+      FC_ASSERT(warning_hs_message, "must provide callback");
+      // no need to std::lock_guard g( _hotstuff_global_mutex ); here since pre-comm init
+      warn_hs_message = std::move(warning_hs_message);
+   }
+
    void chain_pacemaker::get_state(finalizer_state& fs) const {
       // lock-free state version check
       uint64_t current_state_version = _qc_chain.get_state_version();
@@ -297,65 +303,69 @@ namespace eosio { namespace hotstuff {
       prof.core_out();
    }
 
-   void chain_pacemaker::send_hs_proposal_msg(const hs_proposal_message& msg, name id) {
-      bcast_hs_message(msg);
+   void chain_pacemaker::send_hs_proposal_msg(const hs_proposal_message& msg, name id, const std::optional<uint32_t>& exclude_peer) {
+      bcast_hs_message(exclude_peer, msg);
+   }
+
+   void chain_pacemaker::send_hs_vote_msg(const hs_vote_message& msg, name id, const std::optional<uint32_t>& exclude_peer) {
+      bcast_hs_message(exclude_peer, msg);
    }
 
-   void chain_pacemaker::send_hs_vote_msg(const hs_vote_message& msg, name id) {
-      bcast_hs_message(msg);
+   void chain_pacemaker::send_hs_new_block_msg(const hs_new_block_message& msg, name id, const std::optional<uint32_t>& exclude_peer) {
+      bcast_hs_message(exclude_peer, msg);
    }
 
-   void chain_pacemaker::send_hs_new_block_msg(const hs_new_block_message& msg, name id) {
-      bcast_hs_message(msg);
+   void chain_pacemaker::send_hs_new_view_msg(const hs_new_view_message& msg, name id, const std::optional<uint32_t>& exclude_peer) {
+      bcast_hs_message(exclude_peer, msg);
    }
 
-   void chain_pacemaker::send_hs_new_view_msg(const hs_new_view_message& msg, name id) {
-      bcast_hs_message(msg);
+   void chain_pacemaker::send_hs_message_warning(const uint32_t sender_peer, const chain::hs_message_warning code) {
+      warn_hs_message(sender_peer, code);
    }
 
    // called from net threads
-   void chain_pacemaker::on_hs_msg(const eosio::chain::hs_message &msg) {
+   void chain_pacemaker::on_hs_msg(const uint32_t connection_id, const eosio::chain::hs_message &msg) {
       std::visit(overloaded{
-              [this](const hs_vote_message& m) { on_hs_vote_msg(m); },
-              [this](const hs_proposal_message& m) { on_hs_proposal_msg(m); },
-              [this](const hs_new_block_message& m) { on_hs_new_block_msg(m); },
-              [this](const hs_new_view_message& m) { on_hs_new_view_msg(m); },
+            [this, connection_id](const hs_vote_message& m) { on_hs_vote_msg(connection_id, m); },
+            [this, connection_id](const hs_proposal_message& m) { on_hs_proposal_msg(connection_id, m); },
+            [this, connection_id](const hs_new_block_message& m) { on_hs_new_block_msg(connection_id, m); },
+            [this, connection_id](const hs_new_view_message& m) { on_hs_new_view_msg(connection_id, m); },
       }, msg);
    }
 
    // called from net threads
-   void chain_pacemaker::on_hs_proposal_msg(const hs_proposal_message& msg) {
+   void chain_pacemaker::on_hs_proposal_msg(const uint32_t connection_id, const hs_proposal_message& msg) {
       csc prof("prop");
       std::lock_guard g( _hotstuff_global_mutex );
       prof.core_in();
-      _qc_chain.on_hs_proposal_msg(msg);
+      _qc_chain.on_hs_proposal_msg(connection_id, msg);
       prof.core_out();
    }
 
    // called from net threads
-   void chain_pacemaker::on_hs_vote_msg(const hs_vote_message& msg) {
+   void chain_pacemaker::on_hs_vote_msg(const uint32_t connection_id, const hs_vote_message& msg) {
       csc prof("vote");
       std::lock_guard g( _hotstuff_global_mutex );
       prof.core_in();
-      _qc_chain.on_hs_vote_msg(msg);
+      _qc_chain.on_hs_vote_msg(connection_id, msg);
       prof.core_out();
    }
 
    // called from net threads
-   void chain_pacemaker::on_hs_new_block_msg(const hs_new_block_message& msg) {
+   void chain_pacemaker::on_hs_new_block_msg(const uint32_t connection_id, const hs_new_block_message& msg) {
       csc prof("nblk");
       std::lock_guard g( _hotstuff_global_mutex );
       prof.core_in();
-      _qc_chain.on_hs_new_block_msg(msg);
+      _qc_chain.on_hs_new_block_msg(connection_id, msg);
       prof.core_out();
    }
 
    // called from net threads
-   void chain_pacemaker::on_hs_new_view_msg(const hs_new_view_message& msg) {
+   void chain_pacemaker::on_hs_new_view_msg(const uint32_t connection_id, const hs_new_view_message& msg) {
       csc prof("view");
       std::lock_guard g( _hotstuff_global_mutex );
       prof.core_in();
-      _qc_chain.on_hs_new_view_msg(msg);
+      _qc_chain.on_hs_new_view_msg(connection_id, msg);
       prof.core_out();
    }
 

libraries/hotstuff/include/eosio/hotstuff/base_pacemaker.hpp

@@ -2,16 +2,10 @@
 
 #include <eosio/chain/types.hpp>
 #include <eosio/chain/name.hpp>
+#include <eosio/chain/hotstuff.hpp>
 
 #include <vector>
 
-namespace eosio::chain {
-   struct hs_proposal_message;
-   struct hs_vote_message;
-   struct hs_new_view_message;
-   struct hs_new_block_message;
-}
-
 namespace eosio::hotstuff {
 
    // Abstract pacemaker; a reference of this type will only be used by qc_chain, as qc_chain
@@ -36,10 +30,12 @@ namespace eosio::hotstuff {
       virtual std::vector<chain::name> get_finalizers() = 0;
 
       //outbound communications; 'id' is the producer name (can be ignored if/when irrelevant to the implementer)
-      virtual void send_hs_proposal_msg(const chain::hs_proposal_message& msg, chain::name id) = 0;
-      virtual void send_hs_vote_msg(const chain::hs_vote_message& msg, chain::name id) = 0;
-      virtual void send_hs_new_view_msg(const chain::hs_new_view_message& msg, chain::name id) = 0;
-      virtual void send_hs_new_block_msg(const chain::hs_new_block_message& msg, chain::name id) = 0;
+      virtual void send_hs_proposal_msg(const chain::hs_proposal_message& msg, chain::name id, const std::optional<uint32_t>& exclude_peer = std::nullopt) = 0;
+      virtual void send_hs_vote_msg(const chain::hs_vote_message& msg, chain::name id, const std::optional<uint32_t>& exclude_peer = std::nullopt) = 0;
+      virtual void send_hs_new_view_msg(const chain::hs_new_view_message& msg, chain::name id, const std::optional<uint32_t>& exclude_peer = std::nullopt) = 0;
+      virtual void send_hs_new_block_msg(const chain::hs_new_block_message& msg, chain::name id, const std::optional<uint32_t>& exclude_peer = std::nullopt) = 0;
+
+      virtual void send_hs_message_warning(const uint32_t sender_peer, const chain::hs_message_warning code) = 0;
    };
 
 } // namespace eosio::hotstuff

libraries/hotstuff/include/eosio/hotstuff/chain_pacemaker.hpp

@@ -24,11 +24,12 @@ namespace eosio::hotstuff {
                       std::set<account_name> my_producers,
                       chain::bls_key_map_t finalizer_keys,
                       fc::logger& logger);
-      void register_bcast_function(std::function<void(const chain::hs_message&)> broadcast_hs_message);
+      void register_bcast_function(std::function<void(const std::optional<uint32_t>&, const chain::hs_message&)> broadcast_hs_message);
+      void register_warn_function(std::function<void(const uint32_t, const chain::hs_message_warning&)> warning_hs_message);
 
       void beat();
 
-      void on_hs_msg(const hs_message& msg);
+      void on_hs_msg(const uint32_t connection_id, const hs_message& msg);
 
       void get_state(finalizer_state& fs) const;
 
@@ -43,19 +44,21 @@ namespace eosio::hotstuff {
 
       uint32_t get_quorum_threshold();
 
-      void send_hs_proposal_msg(const hs_proposal_message& msg, name id);
-      void send_hs_vote_msg(const hs_vote_message& msg, name id);
-      void send_hs_new_view_msg(const hs_new_view_message& msg, name id);
-      void send_hs_new_block_msg(const hs_new_block_message& msg, name id);
+      void send_hs_proposal_msg(const hs_proposal_message& msg, name id, const std::optional<uint32_t>& exclude_peer);
+      void send_hs_vote_msg(const hs_vote_message& msg, name id, const std::optional<uint32_t>& exclude_peer);
+      void send_hs_new_view_msg(const hs_new_view_message& msg, name id, const std::optional<uint32_t>& exclude_peer);
+      void send_hs_new_block_msg(const hs_new_block_message& msg, name id, const std::optional<uint32_t>& exclude_peer);
+
+      void send_hs_message_warning(const uint32_t sender_peer, const chain::hs_message_warning code);
 
    private:
       void on_accepted_block( const block_state_ptr& blk );
       void on_irreversible_block( const block_state_ptr& blk );
 
-      void on_hs_proposal_msg(const hs_proposal_message& msg); //consensus msg event handler
-      void on_hs_vote_msg(const hs_vote_message& msg); //confirmation msg event handler
-      void on_hs_new_view_msg(const hs_new_view_message& msg); //new view msg event handler
-      void on_hs_new_block_msg(const hs_new_block_message& msg); //new block msg event handler
+      void on_hs_proposal_msg(const uint32_t connection_id, const hs_proposal_message& msg); //consensus msg event handler
+      void on_hs_vote_msg(const uint32_t connection_id, const hs_vote_message& msg); //confirmation msg event handler
+      void on_hs_new_view_msg(const uint32_t connection_id, const hs_new_view_message& msg); //new view msg event handler
+      void on_hs_new_block_msg(const uint32_t connection_id, const hs_new_block_message& msg); //new block msg event handler
    private:
 
       //FIXME/REMOVE: for testing/debugging only
@@ -84,7 +87,8 @@ namespace eosio::hotstuff {
       boost::signals2::scoped_connection _irreversible_block_connection;
 
       qc_chain                _qc_chain;
-      std::function<void(const chain::hs_message&)> bcast_hs_message;
+      std::function<void(const std::optional<uint32_t>&, const chain::hs_message&)> bcast_hs_message;
+      std::function<void(const uint32_t, const chain::hs_message_warning&)> warn_hs_message;
 
       uint32_t                _quorum_threshold = 15; //FIXME/TODO: calculate from schedule
       fc::logger&             _logger;

libraries/hotstuff/include/eosio/hotstuff/qc_chain.hpp

@@ -100,20 +100,20 @@ namespace eosio::hotstuff {
                chain::bls_key_map_t finalizer_keys,
                fc::logger& logger);
 
-      uint64_t get_state_version() const { return _state_version; } // calling this w/ thread sync is optional
+      uint64_t get_state_version() const { return _state_version; } // no lock required
 
-      name get_id_i() const { return _id; } // so far, only ever relevant in a test environment (no sync)
+      name get_id_i() const { return _id; } // only for testing
 
       // Calls to the following methods should be thread-synchronized externally:
 
       void get_state(finalizer_state& fs) const;
 
-      void on_beat(); //handler for pacemaker beat()
+      void on_beat();
 
-      void on_hs_vote_msg(const hs_vote_message& msg); //vote msg event handler
-      void on_hs_proposal_msg(const hs_proposal_message& msg); //proposal msg event handler
-      void on_hs_new_view_msg(const hs_new_view_message& msg); //new view msg event handler
-      void on_hs_new_block_msg(const hs_new_block_message& msg); //new block msg event handler
+      void on_hs_vote_msg(const uint32_t connection_id, const hs_vote_message& msg);
+      void on_hs_proposal_msg(const uint32_t connection_id, const hs_proposal_message& msg);
+      void on_hs_new_view_msg(const uint32_t connection_id, const hs_new_view_message& msg);
+      void on_hs_new_block_msg(const uint32_t connection_id, const hs_new_block_message& msg);
 
    private:
 
@@ -126,48 +126,59 @@ namespace eosio::hotstuff {
 
       hs_bitset update_bitset(const hs_bitset& finalizer_set, name finalizer);
 
-      digest_type get_digest_to_sign(const block_id_type& block_id, uint8_t phase_counter, const fc::sha256& final_on_qc); //get digest to sign from proposal data
+      //get digest to sign from proposal data
+      digest_type get_digest_to_sign(const block_id_type& block_id, uint8_t phase_counter, const fc::sha256& final_on_qc);
 
-      void reset_qc(const fc::sha256& proposal_id); //reset current internal qc
+      void reset_qc(const fc::sha256& proposal_id);
 
-      bool evaluate_quorum(const extended_schedule& es, const hs_bitset& finalizers, const fc::crypto::blslib::bls_signature& agg_sig, const hs_proposal_message& proposal); //evaluate quorum for a proposal
+      //evaluate quorum for a proposal
+      bool evaluate_quorum(const extended_schedule& es, const hs_bitset& finalizers, const fc::crypto::blslib::bls_signature& agg_sig, const hs_proposal_message& proposal);
 
-      // qc.quorum_met has to be updated by the caller (if it wants to) based on the return value of this method
-      bool is_quorum_met(const quorum_certificate& qc, const extended_schedule& schedule, const hs_proposal_message& proposal);  //check if quorum has been met over a proposal
+      //check if quorum has been met over a proposal
+      bool is_quorum_met(const quorum_certificate& qc, const extended_schedule& schedule, const hs_proposal_message& proposal);
 
-      hs_proposal_message new_proposal_candidate(const block_id_type& block_id, uint8_t phase_counter); //create new proposal message
-      hs_new_block_message new_block_candidate(const block_id_type& block_id); //create new block message
+      hs_proposal_message new_proposal_candidate(const block_id_type& block_id, uint8_t phase_counter);
+      hs_new_block_message new_block_candidate(const block_id_type& block_id);
 
-      bool am_i_proposer(); //check if I am the current proposer
-      bool am_i_leader(); //check if I am the current leader
-      bool am_i_finalizer(); //check if I am one of the current finalizers
+      bool am_i_proposer();
+      bool am_i_leader();
+      bool am_i_finalizer();
 
-      void process_proposal(const hs_proposal_message& msg); //handles proposal
-      void process_vote(const hs_vote_message& msg); //handles vote
-      void process_new_view(const hs_new_view_message& msg); //handles new view
-      void process_new_block(const hs_new_block_message& msg); //handles new block
+      // connection_id.has_value() when processing a non-loopback message
+      void process_proposal(const std::optional<uint32_t>& connection_id, const hs_proposal_message& msg);
+      void process_vote(const std::optional<uint32_t>& connection_id, const hs_vote_message& msg);
+      void process_new_view(const std::optional<uint32_t>& connection_id, const hs_new_view_message& msg);
+      void process_new_block(const std::optional<uint32_t>& connection_id, const hs_new_block_message& msg);
 
-      hs_vote_message sign_proposal(const hs_proposal_message& proposal, name finalizer); //sign proposal
+      hs_vote_message sign_proposal(const hs_proposal_message& proposal, name finalizer);
 
-      bool extends(const fc::sha256& descendant, const fc::sha256& ancestor); //verify that a proposal descends from another
+      //verify that a proposal descends from another
+      bool extends(const fc::sha256& descendant, const fc::sha256& ancestor);
 
-      bool update_high_qc(const quorum_certificate& high_qc); //check if update to our high qc is required
+      //update high qc if required
+      bool update_high_qc(const quorum_certificate& high_qc);
 
-      void leader_rotation_check(); //check if leader rotation is required
+      //rotate leader if required
+      void leader_rotation_check();
 
-      bool is_node_safe(const hs_proposal_message& proposal); //verify if a proposal should be signed
+      //verify if a proposal should be signed
+      bool is_node_safe(const hs_proposal_message& proposal);
 
-      std::vector<hs_proposal_message> get_qc_chain(const fc::sha256& proposal_id); //get 3-phase proposal justification
+      //get 3-phase proposal justification
+      std::vector<hs_proposal_message> get_qc_chain(const fc::sha256& proposal_id);
 
-      void send_hs_proposal_msg(const hs_proposal_message& msg); //send vote msg
-      void send_hs_vote_msg(const hs_vote_message& msg); //send proposal msg
-      void send_hs_new_view_msg(const hs_new_view_message& msg); //send new view msg
-      void send_hs_new_block_msg(const hs_new_block_message& msg); //send new block msg
+      // connection_id.has_value() when just propagating a received message
+      void send_hs_proposal_msg(const std::optional<uint32_t>& connection_id, const hs_proposal_message& msg);
+      void send_hs_vote_msg(const std::optional<uint32_t>& connection_id, const hs_vote_message& msg);
+      void send_hs_new_view_msg(const std::optional<uint32_t>& connection_id, const hs_new_view_message& msg);
+      void send_hs_new_block_msg(const std::optional<uint32_t>& connection_id, const hs_new_block_message& msg);
 
-      void update(const hs_proposal_message& proposal); //update internal state
-      void commit(const hs_proposal_message& proposal); //commit proposal (finality)
+      void send_hs_message_warning(const std::optional<uint32_t>& connection_id, const chain::hs_message_warning code);
 
-      void gc_proposals(uint64_t cutoff); //garbage collection of old proposals
+      void update(const hs_proposal_message& proposal);
+      void commit(const hs_proposal_message& proposal);
+
+      void gc_proposals(uint64_t cutoff);
 
 #warning remove. bls12-381 key used for testing purposes
       //todo : remove. bls12-381 key used for testing purposes
@@ -235,7 +246,6 @@ namespace eosio::hotstuff {
 
       proposal_store_type _proposal_store;  //internal proposals store
 #endif
-
    };
 
 } /// eosio::hotstuff

libraries/hotstuff/include/eosio/hotstuff/test_pacemaker.hpp

@@ -57,10 +57,12 @@ namespace eosio { namespace hotstuff {
 
       uint32_t get_quorum_threshold();
 
-      void send_hs_proposal_msg(const hs_proposal_message & msg, name id);
-      void send_hs_vote_msg(const hs_vote_message & msg, name id);
-      void send_hs_new_block_msg(const hs_new_block_message & msg, name id);
-      void send_hs_new_view_msg(const hs_new_view_message & msg, name id);
+      void send_hs_proposal_msg(const hs_proposal_message & msg, name id, const std::optional<uint32_t>& exclude_peer);
+      void send_hs_vote_msg(const hs_vote_message & msg, name id, const std::optional<uint32_t>& exclude_peer);
+      void send_hs_new_block_msg(const hs_new_block_message & msg, name id, const std::optional<uint32_t>& exclude_peer);
+      void send_hs_new_view_msg(const hs_new_view_message & msg, name id, const std::optional<uint32_t>& exclude_peer);
+
+      void send_hs_message_warning(const uint32_t sender_peer, const chain::hs_message_warning code);
 
       std::vector<hotstuff_message> _pending_message_queue;