Issue 2693/pillar votes syncing

libraries/core_libs/consensus/include/pbft/pbft_manager.hpp

@@ -502,6 +502,14 @@ class PbftManager {
   bool validatePbftBlockCertVotes(const std::shared_ptr<PbftBlock> pbft_block,
                                   const std::vector<std::shared_ptr<PbftVote>> &cert_votes) const;
 
+  /**
+   @brief Validates PBFT block [illar] votes
+   *
+   * @param period_data
+   * @return
+   */
+  bool validatePbftBlockPillarVotes(const PeriodData &period_data) const;
+
   /**
    * @param period
    * @return true if node can participate in consensus - is dpos eligible to vote and create blocks for specified period

libraries/core_libs/consensus/include/pillar_chain/pillar_block.hpp

@@ -90,10 +90,14 @@ class PillarBlock {
 };
 
 struct PillarBlockData {
-  std::shared_ptr<PillarBlock> block;
-  std::vector<std::shared_ptr<PillarVote>> pillar_votes;
+  std::shared_ptr<PillarBlock> block_;
+  std::vector<std::shared_ptr<PillarVote>> pillar_votes_;
 
-  HAS_RLP_FIELDS
+  PillarBlockData(std::shared_ptr<PillarBlock> block, std::vector<std::shared_ptr<PillarVote>>&& pillar_votes);
+  PillarBlockData(const dev::RLP& rlp);
+  dev::bytes getRlp() const;
+
+  const static size_t kRlpItemCount = 2;
 };
 
 /** @}*/

libraries/core_libs/consensus/include/pillar_chain/pillar_chain_manager.hpp

@@ -85,9 +85,9 @@ class PillarChainManager {
    * @brief Add a vote to the pillar votes map
    * @param vote vote
    *
-   * @return true if vote was successfully added, otherwise false
+   * @return vote's weight if vote was successfully added, otherwise 0
    */
-  bool addVerifiedPillarVote(const std::shared_ptr<PillarVote>& vote);
+  uint64_t addVerifiedPillarVote(const std::shared_ptr<PillarVote>& vote);
 
   /**
    * @brief Push new finalized pillar block

libraries/core_libs/consensus/src/pbft/pbft_manager.cpp

@@ -928,7 +928,8 @@ void PbftManager::certifyBlock_() {
 
   // Generate pillar vote in case pillar block hash is present in pbft block
   if (const auto pillar_block_hash = soft_voted_block->getPillarBlockHash(); pillar_block_hash.has_value()) {
-    // Creates pillar vote
+    // TODO: validate and save pillar_block_hash in pbft_manager, then try to vote for it during each period if node did
+    // not vote yet, otherwise it might not vote ever, if this block was pushed through syncing Creates pillar vote
     pillar_chain_mgr_->genAndPlacePillarVote(*pillar_block_hash, node_sk_);
   }
 }
@@ -1788,6 +1789,57 @@ std::optional<std::pair<PeriodData, std::vector<std::shared_ptr<PbftVote>>>> Pbf
     }
   }
 
+  // Validate optional pillar block hash
+  const auto kBlockPeriod = period_data.pbft_blk->getPeriod();
+  if (kBlockPeriod > kGenesisConfig.state.hardforks.ficus_hf.pillar_block_periods &&
+      kBlockPeriod % kGenesisConfig.state.hardforks.ficus_hf.pillar_block_periods ==
+          kGenesisConfig.state.dpos.delegation_delay) {
+    if (!period_data.pbft_blk->getPillarBlockHash().has_value()) {
+      LOG(log_er_) << "Synced PBFT block " << pbft_block_hash << ", period " << kBlockPeriod
+                   << " does not contain pillar block hash";
+      sync_queue_.clear();
+      net->handleMaliciousSyncPeer(node_id);
+      return std::nullopt;
+    }
+  } else {
+    if (period_data.pbft_blk->getPillarBlockHash().has_value()) {
+      LOG(log_er_) << "Synced PBFT block " << pbft_block_hash << ", period " << period_data.pbft_blk->getPeriod()
+                   << " contains pillar block hash";
+      sync_queue_.clear();
+      net->handleMaliciousSyncPeer(node_id);
+      return std::nullopt;
+    }
+  }
+
+  // Validate optional pillar votes
+  if (kBlockPeriod >= 2 * kGenesisConfig.state.hardforks.ficus_hf.pillar_block_periods &&
+      kBlockPeriod % kGenesisConfig.state.hardforks.ficus_hf.pillar_block_periods == 0) {
+    if (!period_data.pillar_votes_.has_value()) {
+      LOG(log_er_) << "Synced PBFT block " << pbft_block_hash << ", period " << kBlockPeriod
+                   << " does not contain pillar votes";
+      sync_queue_.clear();
+      net->handleMaliciousSyncPeer(node_id);
+      return std::nullopt;
+    }
+
+    // Validate pillar votes
+    if (!validatePbftBlockPillarVotes(period_data)) {
+      LOG(log_er_) << "Synced PBFT block " << pbft_block_hash << ", period " << kBlockPeriod
+                   << " doesn't have enough valid pillar votes. Clear synced PBFT blocks!";
+      sync_queue_.clear();
+      net->handleMaliciousSyncPeer(node_id);
+      return std::nullopt;
+    }
+  } else {
+    if (period_data.pillar_votes_.has_value()) {
+      LOG(log_er_) << "Synced PBFT block " << pbft_block_hash << ", period " << period_data.pbft_blk->getPeriod()
+                   << " contains pillar votes";
+      sync_queue_.clear();
+      net->handleMaliciousSyncPeer(node_id);
+      return std::nullopt;
+    }
+  }
+
   return std::optional<std::pair<PeriodData, std::vector<std::shared_ptr<PbftVote>>>>(
       {std::move(period_data), std::move(cert_votes)});
 }
@@ -1875,6 +1927,63 @@ bool PbftManager::validatePbftBlockCertVotes(const std::shared_ptr<PbftBlock> pb
   return true;
 }
 
+bool PbftManager::validatePbftBlockPillarVotes(const PeriodData &period_data) const {
+  if (!period_data.pillar_votes_.has_value() || period_data.pillar_votes_->empty()) {
+    LOG(log_er_) << "No pillar votes provided, pbft block period " << period_data.pbft_blk->getPeriod()
+                 << ". The synced PBFT block comes from a malicious player";
+    return false;
+  }
+
+  const auto &pbft_block_hash = period_data.pbft_blk->getBlockHash();
+  const auto kRequiredVotesPeriod =
+      period_data.pbft_blk->getPeriod() - kGenesisConfig.state.hardforks.ficus_hf.pillar_block_periods;
+
+  size_t votes_weight = 0;
+
+  const auto current_pillar_block = pillar_chain_mgr_->getCurrentPillarBlock();
+
+  for (const auto &vote : *period_data.pillar_votes_) {
+    // Any info is wrong that can determine the synced PBFT block comes from a malicious player
+    if (vote->getPeriod() != kRequiredVotesPeriod) {
+      LOG(log_er_) << "Invalid pillar vote " << vote->getHash() << " period " << vote->getPeriod() << ", PBFT block "
+                   << pbft_block_hash << ", kRequiredVotesPeriod " << kRequiredVotesPeriod;
+      return false;
+    }
+
+    if (vote->getBlockHash() != current_pillar_block->getHash()) {
+      LOG(log_er_) << "Invalid pillar vote " << vote->getHash() << ", vote period " << vote->getPeriod()
+                   << ", vote block hash " << vote->getBlockHash() << ", current pillar block "
+                   << current_pillar_block->getHash() << ", block period " << current_pillar_block->getPeriod();
+      return false;
+    }
+
+    if (!pillar_chain_mgr_->validatePillarVote(vote)) {
+      LOG(log_er_) << "Invalid pillar vote " << vote->getHash();
+      return false;
+    }
+
+    if (const auto vote_weight = pillar_chain_mgr_->addVerifiedPillarVote(vote); vote_weight) {
+      votes_weight += vote_weight;
+    } else {
+      LOG(log_er_) << "Unable to add pillar vote " << vote->getHash() << " during syncing";
+      return false;
+    }
+  }
+
+  const auto two_t_plus_one = vote_mgr_->getPbftTwoTPlusOne(kRequiredVotesPeriod - 1, PbftVoteTypes::cert_vote);
+  if (!two_t_plus_one.has_value()) {
+    return false;
+  }
+
+  if (votes_weight < *two_t_plus_one) {
+    LOG(log_wr_) << "Invalid pillar votes weight " << votes_weight << " < two_t_plus_one " << *two_t_plus_one
+                 << ", period " << kRequiredVotesPeriod - 1;
+    return false;
+  }
+
+  return true;
+}
+
 bool PbftManager::canParticipateInConsensus(PbftPeriod period) const {
   try {
     return final_chain_->dpos_is_eligible(period, node_addr_);

libraries/core_libs/consensus/src/pillar_chain/pillar_block.cpp

@@ -4,6 +4,7 @@
 
 #include "common/encoding_rlp.hpp"
 #include "vote/pillar_vote.hpp"
+#include "vote/votes_bundle_rlp.hpp"
 
 namespace taraxa::pillar_chain {
 
@@ -55,6 +56,25 @@ PillarBlock::Hash PillarBlock::getHash() {
 }
 
 RLP_FIELDS_DEFINE(PillarBlock, period_, state_root_, previous_pillar_block_hash_, validators_stakes_changes_)
-RLP_FIELDS_DEFINE(PillarBlockData, block, pillar_votes)
+
+PillarBlockData::PillarBlockData(std::shared_ptr<PillarBlock> block,
+                                 std::vector<std::shared_ptr<PillarVote>>&& pillar_votes)
+    : block_(std::move(block)), pillar_votes_(std::move(pillar_votes)) {}
+PillarBlockData::PillarBlockData(const dev::RLP& rlp) {
+  if (rlp.itemCount() != kRlpItemCount) {
+    throw std::runtime_error("PillarBlockData invalid itemCount: " + std::to_string(rlp.itemCount()));
+  }
+
+  block_ = std::make_shared<PillarBlock>(rlp[0]);
+  pillar_votes_ = decodePillarVotesBundleRlp(rlp[1]);
+}
+
+dev::bytes PillarBlockData::getRlp() const {
+  dev::RLPStream s(kRlpItemCount);
+  s.appendRaw(util::rlp_enc(block_));
+  s.appendRaw(encodePillarVotesBundleRlp(pillar_votes_));
+
+  return s.invalidate();
+}
 
 }  // namespace taraxa::pillar_chain

libraries/core_libs/consensus/src/pillar_chain/pillar_chain_manager.cpp

@@ -32,7 +32,7 @@ PillarChainManager::PillarChainManager(const FicusHardforkConfig& ficusHfConfig,
   }
 
   if (auto&& latest_pillar_block_data = db_->getLatestPillarBlockData(); latest_pillar_block_data.has_value()) {
-    last_finalized_pillar_block_ = std::move(latest_pillar_block_data->block);
+    last_finalized_pillar_block_ = std::move(latest_pillar_block_data->block_);
     // TODO: probably dont need this ???
     //    for (const auto& vote : latest_pillar_block_data->votes) {
     //      addVerifiedPillarVote(vote);
@@ -66,7 +66,7 @@ void PillarChainManager::createPillarBlock(const std::shared_ptr<final_chain::Fi
     }
 
     // Get 2t+1 verified votes
-    const auto two_t_plus_one_votes =
+    auto two_t_plus_one_votes =
         pillar_votes_.getVerifiedVotes(current_pillar_block_->getPeriod(), current_pillar_block_->getHash(), true);
     if (two_t_plus_one_votes.empty()) {
       LOG(log_er_) << "There is < 2t+1 votes for current pillar block " << current_pillar_block_->getHash()
@@ -75,7 +75,7 @@ void PillarChainManager::createPillarBlock(const std::shared_ptr<final_chain::Fi
     }
 
     // Save current pillar block and 2t+1 votes into db
-    if (!pushPillarBlock(PillarBlockData{current_pillar_block_, two_t_plus_one_votes})) {
+    if (!pushPillarBlock(PillarBlockData{current_pillar_block_, std::move(two_t_plus_one_votes)})) {
       // This should never happen
       LOG(log_er_) << "Unable to push pillar block: " << current_pillar_block_->getHash() << ", period "
                    << current_pillar_block_->getPeriod();
@@ -163,18 +163,18 @@ bool PillarChainManager::genAndPlacePillarVote(const PillarBlock::Hash& pillar_b
 
 bool PillarChainManager::pushPillarBlock(const PillarBlockData& pillarBlockData) {
   // Note: 2t+1 votes should be validated before calling pushPillarBlock
-  if (!isValidPillarBlock(pillarBlockData.block)) {
+  if (!isValidPillarBlock(pillarBlockData.block_)) {
     LOG(log_er_) << "Trying to push invalid pillar block";
     return false;
   }
 
   db_->savePillarBlockData(pillarBlockData);
-  LOG(log_nf_) << "Pillar block " << pillarBlockData.block->getHash() << " with period "
-               << pillarBlockData.block->getPeriod() << " pushed into the pillar chain";
+  LOG(log_nf_) << "Pillar block " << pillarBlockData.block_->getHash() << " with period "
+               << pillarBlockData.block_->getPeriod() << " pushed into the pillar chain";
 
   {
     std::scoped_lock<std::shared_mutex> lock(mutex_);
-    last_finalized_pillar_block_ = pillarBlockData.block;
+    last_finalized_pillar_block_ = pillarBlockData.block_;
 
     // Erase votes that are no longer needed
     pillar_votes_.eraseVotes(last_finalized_pillar_block_->getPeriod());
@@ -279,20 +279,20 @@ bool PillarChainManager::validatePillarVote(const std::shared_ptr<PillarVote> vo
   return true;
 }
 
-bool PillarChainManager::addVerifiedPillarVote(const std::shared_ptr<PillarVote>& vote) {
+uint64_t PillarChainManager::addVerifiedPillarVote(const std::shared_ptr<PillarVote>& vote) {
   uint64_t validator_vote_count = 0;
   try {
     validator_vote_count = final_chain_->dpos_eligible_vote_count(vote->getPeriod(), vote->getVoterAddr());
   } catch (state_api::ErrFutureBlock& e) {
     LOG(log_er_) << "Pillar vote " << vote->getHash() << " with period " << vote->getPeriod()
                  << " is too far ahead of DPOS. " << e.what();
-    return false;
+    return 0;
   }
 
   if (!validator_vote_count) {
     LOG(log_er_) << "Zero stake for pillar vote: " << vote->getHash() << ", author: " << vote->getVoterAddr()
                  << ", period: " << vote->getPeriod();
-    return false;
+    return 0;
   }
 
   if (!pillar_votes_.periodDataInitialized(vote->getPeriod())) {
@@ -306,18 +306,18 @@ bool PillarChainManager::addVerifiedPillarVote(const std::shared_ptr<PillarVote>
       // happen as this exception is caught above when calling dpos_eligible_vote_count
       LOG(log_er_) << "Unable to get 2t+1 for period " << vote->getPeriod();
       assert(false);
-      return false;
+      return 0;
     }
   }
 
   if (!pillar_votes_.addVerifiedVote(vote, validator_vote_count)) {
     LOG(log_er_) << "Non-unique pillar vote " << vote->getHash() << ", validator " << vote->getVoterAddr();
-    return false;
+    return 0;
   }
 
   LOG(log_dg_) << "Pillar vote " << vote->getHash() << " with period " << vote->getPeriod() << " for block "
                << vote->getBlockHash() << " added to the verified votes";
-  return true;
+  return validator_vote_count;
 }
 
 std::vector<std::shared_ptr<PillarVote>> PillarChainManager::getVerifiedPillarVotes(

libraries/core_libs/consensus/src/pillar_chain/pillar_votes.cpp

@@ -3,6 +3,8 @@
 namespace taraxa::pillar_chain {
 
 bool PillarVotes::voteExists(const std::shared_ptr<PillarVote> vote) const {
+  std::shared_lock<std::shared_mutex> lock(mutex_);
+
   const auto found_period_votes = votes_.find(vote->getPeriod());
   if (found_period_votes == votes_.end()) {
     return false;

libraries/core_libs/network/include/network/tarcap/packets_handlers/latest/get_pbft_sync_packet_handler.hpp

@@ -8,6 +8,10 @@ class DbStorage;
 class VoteManager;
 }  // namespace taraxa
 
+namespace taraxa::pillar_chain {
+class PillarChainManager;
+}
+
 namespace taraxa::network::tarcap {
 
 class PbftSyncingState;
@@ -17,8 +21,10 @@ class GetPbftSyncPacketHandler : public PacketHandler {
   GetPbftSyncPacketHandler(const FullNodeConfig& conf, std::shared_ptr<PeersState> peers_state,
                            std::shared_ptr<TimePeriodPacketsStats> packets_stats,
                            std::shared_ptr<PbftSyncingState> pbft_syncing_state, std::shared_ptr<PbftChain> pbft_chain,
-                           std::shared_ptr<VoteManager> vote_mgr, std::shared_ptr<DbStorage> db,
-                           const addr_t& node_addr, const std::string& logs_prefix = "GET_PBFT_SYNC_PH");
+                           std::shared_ptr<VoteManager> vote_mgr,
+                           std::shared_ptr<pillar_chain::PillarChainManager> pillar_chain_mgr,
+                           std::shared_ptr<DbStorage> db, const addr_t& node_addr,
+                           const std::string& logs_prefix = "GET_PBFT_SYNC_PH");
 
   // Packet type that is processed by this handler
   static constexpr SubprotocolPacketType kPacketType_ = SubprotocolPacketType::GetPbftSyncPacket;
@@ -34,6 +40,7 @@ class GetPbftSyncPacketHandler : public PacketHandler {
   std::shared_ptr<PbftSyncingState> pbft_syncing_state_;
   std::shared_ptr<PbftChain> pbft_chain_;
   std::shared_ptr<VoteManager> vote_mgr_;
+  std::shared_ptr<pillar_chain::PillarChainManager> pillar_chain_mgr_;
   std::shared_ptr<DbStorage> db_;
 };
 

libraries/core_libs/network/src/tarcap/packets_handlers/latest/common/ext_votes_packet_handler.cpp

@@ -182,7 +182,7 @@ void ExtVotesPacketHandler::sendPbftVotesBundle(const std::shared_ptr<TaraxaPeer
   }
 
   auto sendVotes = [this, &peer](std::vector<std::shared_ptr<PbftVote>> &&votes) {
-    auto votes_bytes = encodeVotesBundleRlp(std::move(votes), false);
+    auto votes_bytes = encodePbftVotesBundleRlp(std::move(votes));
     if (votes_bytes.empty()) {
       LOG(log_er_) << "Unable to send VotesBundle rlp";
       return;