Ordering Service: on request proposal strategy

irohad/consensus/CMakeLists.txt

@@ -8,6 +8,7 @@ add_library(consensus_round
     )
 target_link_libraries(consensus_round
     boost
+    shared_model_utils
     )
 
 add_library(gate_object

irohad/consensus/yac/impl/supermajority_checker_bft.cpp

@@ -19,6 +19,11 @@ namespace iroha {
             agreed, all, detail::kSupermajorityCheckerKfPlus1Bft);
       }
 
+      bool SupermajorityCheckerBft::hasMajority(PeersNumberType voted,
+                       PeersNumberType all) const {
+        return checkKfPlus1Supermajority(voted, all, 2);
+      }
+
       bool SupermajorityCheckerBft::canHaveSupermajority(
           const VoteGroups &votes, PeersNumberType all) const {
         const PeersNumberType largest_group =

irohad/consensus/yac/impl/supermajority_checker_bft.hpp

@@ -23,6 +23,9 @@ namespace iroha {
         bool hasSupermajority(PeersNumberType current,
                               PeersNumberType all) const override;
 
+        bool hasMajority(PeersNumberType voted,
+                         PeersNumberType all) const override;
+
         bool canHaveSupermajority(const VoteGroups &votes,
                                   PeersNumberType all) const override;
       };

irohad/consensus/yac/impl/supermajority_checker_cft.cpp

@@ -19,6 +19,11 @@ namespace iroha {
             agreed, all, detail::kSupermajorityCheckerKfPlus1Cft);
       }
 
+      bool SupermajorityCheckerCft::hasMajority(PeersNumberType voted,
+                       PeersNumberType all) const{
+        return hasSupermajority(voted, all);
+      }
+
       bool SupermajorityCheckerCft::canHaveSupermajority(
           const VoteGroups &votes, PeersNumberType all) const {
         const PeersNumberType largest_group =

irohad/consensus/yac/impl/supermajority_checker_cft.hpp

@@ -23,6 +23,9 @@ namespace iroha {
         bool hasSupermajority(PeersNumberType current,
                               PeersNumberType all) const override;
 
+        bool hasMajority(PeersNumberType voted,
+                         PeersNumberType all) const override;
+
         bool canHaveSupermajority(const VoteGroups &votes,
                                   PeersNumberType all) const override;
       };

irohad/consensus/yac/supermajority_checker.hpp

@@ -49,6 +49,15 @@ namespace iroha {
         virtual bool hasSupermajority(PeersNumberType current,
                                       PeersNumberType all) const = 0;
 
+        /**
+         * Check if majority of votes is achieved
+         * @param voted - number of voted peers
+         * @param all - number of all peers in network
+         * @return true if majority is reached
+         */
+        virtual bool hasMajority(PeersNumberType voted,
+                                 PeersNumberType all) const = 0;
+
         /**
          * Check if supermajority is possible
          * @param voted - numbers of peers voted for each option

irohad/main/application.cpp

@@ -35,6 +35,7 @@
 #include "multi_sig_transactions/transport/mst_transport_stub.hpp"
 #include "network/impl/block_loader_impl.hpp"
 #include "network/impl/peer_communication_service_impl.hpp"
+#include "ordering/impl/kick_out_proposal_creation_strategy.hpp"
 #include "ordering/impl/on_demand_common.hpp"
 #include "ordering/impl/on_demand_ordering_gate.hpp"
 #include "pending_txs_storage/impl/pending_txs_storage_impl.hpp"
@@ -374,6 +375,15 @@ void Irohad::initOrderingGate() {
     return reject_delay;
   };
 
+  std::shared_ptr<iroha::ordering::ProposalCreationStrategy> proposal_strategy =
+      std::make_shared<ordering::KickOutProposalCreationStrategy>(
+          getSupermajorityChecker(kConsensusConsistencyModel));
+
+  auto field_validator =
+      std::make_shared<shared_model::validation::FieldValidator>();
+  auto self_peer_key =
+      std::make_shared<shared_model::crypto::PublicKey>(keypair.publicKey());
+
   ordering_gate =
       ordering_init.initOrderingGate(max_proposal_size_,
                                      proposal_delay_,
@@ -387,7 +397,10 @@ void Irohad::initOrderingGate() {
                                      proposal_factory,
                                      persistent_cache,
                                      delay,
-                                     log_manager_->getChild("Ordering"));
+                                     log_manager_->getChild("Ordering"),
+                                     field_validator,
+                                     proposal_strategy,
+                                     self_peer_key);
   log_->info("[Init] => init ordering gate - [{}]",
              logger::logBool(ordering_gate));
 }
@@ -679,6 +692,21 @@ Irohad::RunResult Irohad::run() {
                                          + e->error);
             }
 
+            auto peer_query = storage->createPeerQuery();
+            if (not peer_query) {
+              return expected::makeError("Failed to create peer query");
+            }
+            auto peer_list = (*peer_query)->getLedgerPeers();
+            if (not peer_list) {
+              return expected::makeError(
+                  "Failed to fetch peers from peer query");
+            }
+            std::shared_ptr<iroha::LedgerState> state =
+                std::make_shared<iroha::LedgerState>();
+            state->ledger_peers = std::make_shared<
+                std::vector<std::shared_ptr<shared_model::interface::Peer>>>(
+                std::move(peer_list.get()));
+
             auto block = boost::get<expected::Value<
                 std::shared_ptr<shared_model::interface::Block>>>(&block_var)
                              ->value;
@@ -689,7 +717,8 @@ Irohad::RunResult Irohad::run() {
                 synchronizer::SynchronizationEvent{
                     rxcpp::observable<>::just(block),
                     SynchronizationOutcomeType::kCommit,
-                    {block->height(), ordering::kFirstRejectRound}});
+                    {block->height(), ordering::kFirstRejectRound},
+                    state});
             return {};
           },
           [&](const expected::Error<std::string> &e) -> RunResult {

irohad/main/impl/on_demand_ordering_init.cpp

@@ -13,6 +13,7 @@
 #include "interfaces/common_objects/types.hpp"
 #include "logger/logger.hpp"
 #include "logger/logger_manager.hpp"
+#include "ordering/impl/kick_out_proposal_creation_strategy.hpp"
 #include "ordering/impl/on_demand_common.hpp"
 #include "ordering/impl/on_demand_connection_manager.hpp"
 #include "ordering/impl/on_demand_ordering_gate.hpp"
@@ -51,12 +52,14 @@ namespace iroha {
             async_call,
         std::shared_ptr<TransportFactoryType> proposal_transport_factory,
         std::chrono::milliseconds delay,
+        std::shared_ptr<shared_model::crypto::PublicKey> self_peer,
         const logger::LoggerManagerTreePtr &ordering_log_manager) {
       return std::make_shared<ordering::transport::OnDemandOsClientGrpcFactory>(
           std::move(async_call),
           std::move(proposal_transport_factory),
           [] { return std::chrono::system_clock::now(); },
           delay,
+          self_peer,
           ordering_log_manager->getChild("NetworkClient")->getLogger());
     }
 
@@ -67,6 +70,7 @@ namespace iroha {
         std::shared_ptr<TransportFactoryType> proposal_transport_factory,
         std::chrono::milliseconds delay,
         std::vector<shared_model::interface::types::HashType> initial_hashes,
+        std::shared_ptr<shared_model::crypto::PublicKey> self_peer_key,
         const logger::LoggerManagerTreePtr &ordering_log_manager) {
       // since top block will be the first in notifier observable, hashes of
       // two previous blocks are prepended
@@ -194,6 +198,7 @@ namespace iroha {
           createNotificationFactory(std::move(async_call),
                                     std::move(proposal_transport_factory),
                                     delay,
+                                    self_peer_key,
                                     ordering_log_manager),
           peers,
           ordering_log_manager->getChild("ConnectionManager")->getLogger());
@@ -232,12 +237,15 @@ namespace iroha {
                               std::inserter(hashes, hashes.end()));
                   });
               return ordering::OnDemandOrderingGate::BlockEvent{
-                  ordering::nextCommitRound(commit.round), hashes};
+                  ordering::nextCommitRound(commit.round),
+                  hashes,
+                  commit.ledger_state};
             },
             [](const auto &nothing)
                 -> ordering::OnDemandOrderingGate::BlockRoundEventType {
               return ordering::OnDemandOrderingGate::EmptyEvent{
-                  ordering::nextRejectRound(nothing.round)};
+                  ordering::nextRejectRound(nothing.round),
+                  nothing.ledger_state};
             });
       };
 
@@ -261,11 +269,13 @@ namespace iroha {
         std::shared_ptr<shared_model::interface::UnsafeProposalFactory>
             proposal_factory,
         std::shared_ptr<ametsuchi::TxPresenceCache> tx_cache,
+        std::shared_ptr<ordering::ProposalCreationStrategy> creation_strategy,
         const logger::LoggerManagerTreePtr &ordering_log_manager) {
       return std::make_shared<ordering::OnDemandOrderingServiceImpl>(
           max_number_of_transactions,
           std::move(proposal_factory),
           std::move(tx_cache),
+          creation_strategy,
           ordering_log_manager->getChild("Service")->getLogger());
     }
 
@@ -294,16 +304,22 @@ namespace iroha {
         std::shared_ptr<ametsuchi::TxPresenceCache> tx_cache,
         std::function<std::chrono::milliseconds(
             const synchronizer::SynchronizationEvent &)> delay_func,
-        logger::LoggerManagerTreePtr ordering_log_manager) {
+        logger::LoggerManagerTreePtr ordering_log_manager,
+        std::shared_ptr<shared_model::validation::FieldValidator>
+            field_validator,
+        std::shared_ptr<ordering::ProposalCreationStrategy> creation_strategy,
+        std::shared_ptr<shared_model::crypto::PublicKey> self_peer_key) {
       auto ordering_service = createService(max_number_of_transactions,
                                             proposal_factory,
                                             tx_cache,
+                                            creation_strategy,
                                             ordering_log_manager);
       service = std::make_shared<ordering::transport::OnDemandOsServerGrpc>(
           ordering_service,
           std::move(transaction_factory),
           std::move(batch_parser),
           std::move(transaction_batch_factory),
+          field_validator,
           ordering_log_manager->getChild("Server")->getLogger());
       return createGate(
           ordering_service,
@@ -312,6 +328,7 @@ namespace iroha {
                                   std::move(proposal_transport_factory),
                                   delay,
                                   std::move(initial_hashes),
+                                  self_peer_key,
                                   ordering_log_manager),
           std::make_shared<ordering::cache::OnDemandCache>(),
           std::move(proposal_factory),

irohad/main/impl/on_demand_ordering_init.hpp

@@ -20,7 +20,7 @@
 #include "ordering/impl/on_demand_os_server_grpc.hpp"
 #include "ordering/impl/ordering_gate_cache/ordering_gate_cache.hpp"
 #include "ordering/on_demand_ordering_service.hpp"
-#include "ordering/on_demand_os_transport.hpp"
+#include "ordering/ordering_service_proposal_creation_strategy.hpp"
 
 namespace iroha {
   namespace network {
@@ -45,6 +45,7 @@ namespace iroha {
               async_call,
           std::shared_ptr<TransportFactoryType> proposal_transport_factory,
           std::chrono::milliseconds delay,
+          std::shared_ptr<shared_model::crypto::PublicKey> self_peer,
           const logger::LoggerManagerTreePtr &ordering_log_manager);
 
       /**
@@ -59,6 +60,7 @@ namespace iroha {
           std::shared_ptr<TransportFactoryType> proposal_transport_factory,
           std::chrono::milliseconds delay,
           std::vector<shared_model::interface::types::HashType> initial_hashes,
+          std::shared_ptr<shared_model::crypto::PublicKey> self_peer,
           const logger::LoggerManagerTreePtr &ordering_log_manager);
 
       /**
@@ -86,6 +88,7 @@ namespace iroha {
           std::shared_ptr<shared_model::interface::UnsafeProposalFactory>
               proposal_factory,
           std::shared_ptr<ametsuchi::TxPresenceCache> tx_cache,
+          std::shared_ptr<ordering::ProposalCreationStrategy> creation_strategy,
           const logger::LoggerManagerTreePtr &ordering_log_manager);
 
      public:
@@ -98,7 +101,8 @@ namespace iroha {
       /**
        * Initializes on-demand ordering gate and ordering sevice components
        *
-       * @param max_number_of_transactions maximum number of transaction in a proposal
+       * @param max_number_of_transactions maximum number of transaction in a
+       * proposal
        * @param delay timeout for ordering service response on proposal request
        * @param initial_hashes seeds for peer list permutations for first k
        * rounds they are required since hash of block i defines round i + k
@@ -114,6 +118,10 @@ namespace iroha {
        * requests to ordering service and processing responses
        * @param proposal_factory factory required by ordering service to produce
        * proposals
+       * @param field_validator - provides a dependency for validating peer keys
+       * @param creation_strategy - provides a strategy for creaing proposals in
+       * OS
+       * @param self_peer_key - the public key of instantiated peer
        * @return initialized ordering gate
        */
       std::shared_ptr<network::OrderingGate> initOrderingGate(
@@ -136,7 +144,11 @@ namespace iroha {
           std::shared_ptr<ametsuchi::TxPresenceCache> tx_cache,
           std::function<std::chrono::milliseconds(
               const synchronizer::SynchronizationEvent &)> delay_func,
-          logger::LoggerManagerTreePtr ordering_log_manager);
+          logger::LoggerManagerTreePtr ordering_log_manager,
+          std::shared_ptr<shared_model::validation::FieldValidator>
+              field_validator,
+          std::shared_ptr<ordering::ProposalCreationStrategy> creation_strategy,
+          std::shared_ptr<shared_model::crypto::PublicKey> self_peer_key);
 
       /// gRPC service for ordering service
       std::shared_ptr<ordering::proto::OnDemandOrdering::Service> service;

irohad/ordering/CMakeLists.txt

@@ -11,6 +11,7 @@ target_link_libraries(on_demand_common
 
 add_library(on_demand_ordering_service
     impl/on_demand_ordering_service_impl.cpp
+    impl/kick_out_proposal_creation_strategy.cpp
     )
 
 target_link_libraries(on_demand_ordering_service

irohad/ordering/impl/kick_out_proposal_creation_strategy.cpp

@@ -0,0 +1,60 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "ordering/impl/kick_out_proposal_creation_strategy.hpp"
+
+#include <numeric>
+#include "interfaces/common_objects/peer.hpp"
+
+using namespace iroha::ordering;
+
+KickOutProposalCreationStrategy::KickOutProposalCreationStrategy(
+    std::shared_ptr<SupermajorityCheckerType> majority_checker)
+    : majority_checker_(majority_checker) {}
+
+void KickOutProposalCreationStrategy::onCollaborationOutcome(
+    const PeerList &peers) {
+  last_requested_ =
+      std::accumulate(peers.begin(),
+                      peers.end(),
+                      RoundCollectionType{},
+                      [this](auto collection, const auto &peer) {
+                        auto iter = last_requested_.find(peer->hex());
+                        if (iter != last_requested_.end()) {
+                          collection.insert(*iter);
+                        } else {
+                          collection.insert({peer->hex(), RoundType{0, 0}});
+                        }
+                        return std::move(collection);
+                      });
+}
+
+bool KickOutProposalCreationStrategy::shouldCreateRound(RoundType round) {
+  uint64_t counter = 0;
+  std::for_each(last_requested_.begin(),
+                last_requested_.end(),
+                [&round, &counter](const auto &elem) {
+                  if (elem.second >= round) {
+                    counter++;
+                  }
+                });
+
+  auto has_majority =
+      majority_checker_->hasMajority(counter, last_requested_.size());
+  return not has_majority;
+}
+
+boost::optional<ProposalCreationStrategy::RoundType>
+KickOutProposalCreationStrategy::onProposal(PeerType who,
+                                            RoundType requested_round) {
+  auto iter = last_requested_.find(who->hex());
+  if (iter != last_requested_.end()) {
+    if (iter->second < requested_round) {
+      iter->second = requested_round;
+    }
+  }
+
+  return boost::none;
+}