implement pillar votes syncing as part of pbft syncing

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

@@ -90,9 +90,10 @@ 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_;
 
+  PillarBlockData(std::shared_ptr<PillarBlock> block, std::vector<std::shared_ptr<PillarVote>>&& pillar_votes);
   PillarBlockData(const dev::RLP& rlp);
   dev::bytes getRlp() const;
 

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

@@ -1928,8 +1928,60 @@ bool PbftManager::validatePbftBlockCertVotes(const std::shared_ptr<PbftBlock> pb
 }
 
 bool PbftManager::validatePbftBlockPillarVotes(const PeriodData &period_data) const {
-  // TODO: implement validation - similar to validatePbftBlockCertVotes
-  return false;
+  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 {

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

@@ -57,19 +57,22 @@ PillarBlock::Hash PillarBlock::getHash() {
 
 RLP_FIELDS_DEFINE(PillarBlock, period_, state_root_, previous_pillar_block_hash_, validators_stakes_changes_)
 
+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]);
+  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));
+  s.appendRaw(util::rlp_enc(block_));
+  s.appendRaw(encodePillarVotesBundleRlp(pillar_votes_));
 
   return s.invalidate();
 }

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/network/include/network/tarcap/packets_handlers/latest/get_pbft_sync_packet_handler.hpp

@@ -6,9 +6,12 @@ namespace taraxa {
 class PbftChain;
 class DbStorage;
 class VoteManager;
-class PillarChainManager;
 }  // namespace taraxa
 
+namespace taraxa::pillar_chain {
+class PillarChainManager;
+}
+
 namespace taraxa::network::tarcap {
 
 class PbftSyncingState;
@@ -18,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<PillarChainManager> pillar_chain_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;
@@ -35,7 +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<PillarChainManager> pillar_chain_mgr_;
+  std::shared_ptr<pillar_chain::PillarChainManager> pillar_chain_mgr_;
   std::shared_ptr<DbStorage> db_;
 };
 

libraries/core_libs/network/src/tarcap/packets_handlers/latest/get_pbft_sync_packet_handler.cpp

@@ -2,6 +2,7 @@
 
 #include "network/tarcap/shared_states/pbft_syncing_state.hpp"
 #include "pbft/pbft_chain.hpp"
+#include "pillar_chain/pillar_chain_manager.hpp"
 #include "storage/storage.hpp"
 #include "vote/pbft_vote.hpp"
 #include "vote/votes_bundle_rlp.hpp"
@@ -13,9 +14,10 @@ GetPbftSyncPacketHandler::GetPbftSyncPacketHandler(const FullNodeConfig &conf, s
                                                    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<PillarChainManager> pillar_chain_mgr,
-                                                   std::shared_ptr<DbStorage> db,
-                                                   const addr_t &node_addr, const std::string &logs_prefix)
+                                                   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)
     : PacketHandler(conf, std::move(peers_state), std::move(packets_stats), node_addr,
                     logs_prefix + "GET_PBFT_SYNC_PH"),
       pbft_syncing_state_(std::move(pbft_syncing_state)),
@@ -83,14 +85,19 @@ void GetPbftSyncPacketHandler::sendPbftBlocks(const std::shared_ptr<TaraxaPeer>
       return;
     }
 
+    // TODO: bad solution: should not decode PeriodData, add pillar votes and then encode it...
+    PeriodData period_data{data};
     // Add pillar votes to period data
     if (block_period >= 2 * kConf.genesis.state.hardforks.ficus_hf.pillar_block_periods &&
         block_period % kConf.genesis.state.hardforks.ficus_hf.pillar_block_periods == 0) {
-      const auto pillar_votes = db_->getPillarBlockData(block_period - kConf.genesis.state.hardforks.ficus_hf.pillar_block_periods);
-      if (!pillar_votes.has_value()) {
+      auto pillar_data =
+          db_->getPillarBlockData(block_period - kConf.genesis.state.hardforks.ficus_hf.pillar_block_periods);
+      if (!pillar_data.has_value()) {
         LOG(log_er_) << "DB corrupted. Cannot find pillar votes for period " << block_period << " in db";
         return;
       }
+
+      period_data.pillar_votes_ = std::move(pillar_data->pillar_votes_);
     }
 
     dev::RLPStream s;
@@ -102,17 +109,17 @@ void GetPbftSyncPacketHandler::sendPbftBlocks(const std::shared_ptr<TaraxaPeer>
       if (reward_votes[0]->getPeriod() == block_period) {
         s.appendList(3);
         s << last_block;
-        s.appendRaw(data);
+        s.appendRaw(period_data.rlp());
         s.appendRaw(encodePbftVotesBundleRlp(reward_votes));
       } else {
         s.appendList(2);
         s << last_block;
-        s.appendRaw(data);
+        s.appendRaw(period_data.rlp());
       }
     } else {
       s.appendList(2);
       s << last_block;
-      s.appendRaw(data);
+      s.appendRaw(period_data.rlp());
     }
     LOG(log_dg_) << "Sending PbftSyncPacket period " << block_period << " to " << peer_id;
     sealAndSend(peer_id, SubprotocolPacketType::PbftSyncPacket, std::move(s));

libraries/core_libs/network/src/tarcap/packets_handlers/latest/get_pillar_chain_sync_packet_handler.cpp

@@ -45,7 +45,7 @@ void GetPillarChainSyncPacketHandler::process(const threadpool::PacketData &pack
     }
 
     dev::RLPStream s;
-    s.appendRaw(util::rlp_enc(*pillar_block_data));
+    //    s.appendRaw(util::rlp_enc(*pillar_block_data));
 
     LOG(log_dg_) << "Sending PillarChainSyncPacket for period " << period << " to " << peer_id;
     sealAndSend(peer_id, SubprotocolPacketType::PillarChainSyncPacket, std::move(s));

libraries/core_libs/network/src/tarcap/packets_handlers/latest/pbft_sync_packet_handler.cpp

@@ -131,6 +131,43 @@ void PbftSyncPacketHandler::process(const threadpool::PacketData &packet_data,
       }
     }
 
+    // Validate optional pillar block hash
+    const auto kFicusHfConfig = kConf.genesis.state.hardforks.ficus_hf;
+    if (pbft_block_period > kFicusHfConfig.pillar_block_periods &&
+        pbft_block_period % kFicusHfConfig.pillar_block_periods == kConf.genesis.state.dpos.delegation_delay) {
+      if (!period_data.pbft_blk->getPillarBlockHash().has_value()) {
+        LOG(log_er_) << "Synced PBFT block " << pbft_blk_hash << ", period " << pbft_block_period
+                     << " does not contain pillar block hash";
+        handleMaliciousSyncPeer(packet_data.from_node_id_);
+        return;
+      }
+    } else {
+      if (period_data.pbft_blk->getPillarBlockHash().has_value()) {
+        LOG(log_er_) << "Synced PBFT block " << pbft_blk_hash << ", period " << period_data.pbft_blk->getPeriod()
+                     << " contains pillar block hash";
+        handleMaliciousSyncPeer(packet_data.from_node_id_);
+        return;
+      }
+    }
+
+    // Validate optional pillar votes
+    if (pbft_block_period >= 2 * kFicusHfConfig.pillar_block_periods &&
+        pbft_block_period % kFicusHfConfig.pillar_block_periods == 0) {
+      if (!period_data.pillar_votes_.has_value()) {
+        LOG(log_er_) << "Synced PBFT block " << pbft_blk_hash << ", period " << pbft_block_period
+                     << " does not contain pillar votes";
+        handleMaliciousSyncPeer(packet_data.from_node_id_);
+        return;
+      }
+    } else {
+      if (period_data.pillar_votes_.has_value()) {
+        LOG(log_er_) << "Synced PBFT block " << pbft_blk_hash << ", period " << period_data.pbft_blk->getPeriod()
+                     << " contains pillar votes";
+        handleMaliciousSyncPeer(packet_data.from_node_id_);
+        return;
+      }
+    }
+
     auto order_hash = PbftManager::calculateOrderHash(period_data.dag_blocks);
     if (order_hash != period_data.pbft_blk->getOrderHash()) {
       {  // This is just log related stuff