Fix itf fake peers on demand ordering test

test/framework/integration_framework/fake_peer/behaviour/behaviour.cpp

@@ -21,29 +21,56 @@ namespace integration_framework {
       // Hint: such calls would precede the derived class construction.
       fake_peer_wptr_ = fake_peer;
       log_ = std::move(log);
+      std::weak_ptr<Behaviour> weak_this = shared_from_this();
       // subscribe for all messages
       subscriptions_.emplace_back(
           getFakePeer().getMstStatesObservable().subscribe(
-              [this](const auto &message) {
-                this->processMstMessage(message);
+              [weak_this,
+               weak_fake_peer = fake_peer_wptr_](const auto &message) {
+                auto me_alive = weak_this.lock();
+                auto fake_peer_is_alive = weak_fake_peer.lock();
+                if (me_alive and fake_peer_is_alive) {
+                  me_alive->processMstMessage(message);
+                }
               }));
       subscriptions_.emplace_back(
           getFakePeer().getYacStatesObservable().subscribe(
-              [this](const auto &message) {
-                this->processYacMessage(message);
+              [weak_this,
+               weak_fake_peer = fake_peer_wptr_](const auto &message) {
+                auto me_alive = weak_this.lock();
+                auto fake_peer_is_alive = weak_fake_peer.lock();
+                if (me_alive and fake_peer_is_alive) {
+                  me_alive->processYacMessage(message);
+                }
               }));
       subscriptions_.emplace_back(
           getFakePeer().getOsBatchesObservable().subscribe(
-              [this](const auto &batch) { this->processOsBatch(batch); }));
+              [weak_this, weak_fake_peer = fake_peer_wptr_](const auto &batch) {
+                auto me_alive = weak_this.lock();
+                auto fake_peer_is_alive = weak_fake_peer.lock();
+                if (me_alive and fake_peer_is_alive) {
+                  me_alive->processOsBatch(batch);
+                }
+              }));
       subscriptions_.emplace_back(
           getFakePeer().getOgProposalsObservable().subscribe(
-              [this](const auto &proposal) {
-                this->processOgProposal(proposal);
+              [weak_this,
+               weak_fake_peer = fake_peer_wptr_](const auto &proposal) {
+                auto me_alive = weak_this.lock();
+                auto fake_peer_is_alive = weak_fake_peer.lock();
+                if (me_alive and fake_peer_is_alive) {
+                  me_alive->processOgProposal(proposal);
+                }
               }));
       subscriptions_.emplace_back(
           getFakePeer().getBatchesObservable().subscribe(
-              [this](const auto &batches) {
-                this->processOrderingBatches(*batches);
+              [weak_this,
+               weak_fake_peer = fake_peer_wptr_](const auto &batches) {
+                auto me_alive = weak_this.lock();
+                auto fake_peer_is_alive = weak_fake_peer.lock();
+                if (me_alive and fake_peer_is_alive) {
+                  me_alive->processOrderingBatches(*batches);
+                }
               }));
     }
 

test/framework/integration_framework/fake_peer/behaviour/behaviour.hpp

@@ -23,7 +23,7 @@ namespace shared_model {
 namespace integration_framework {
   namespace fake_peer {
 
-    class Behaviour {
+    class Behaviour : public std::enable_shared_from_this<Behaviour> {
      public:
       virtual ~Behaviour();
 

test/framework/integration_framework/fake_peer/behaviour/honest.cpp

@@ -77,9 +77,9 @@ namespace integration_framework {
       auto opt_proposal = proposal_storage->getProposal(request);
       getLogger()->debug(
           "Got an OnDemandOrderingService.GetProposal call for round {}, "
-          "{} returning a proposal.",
+          "{}returning a proposal.",
           request.toString(),
-          opt_proposal ? "" : "NOT");
+          opt_proposal ? "" : "NOT ");
       return opt_proposal;
     }
 

test/framework/integration_framework/fake_peer/fake_peer.cpp

@@ -124,6 +124,12 @@ namespace integration_framework {
       yac_transport_->subscribe(yac_network_notifier_);
     }
 
+    FakePeer::~FakePeer() {
+      if (behaviour_) {
+        behaviour_->absolve();
+      }
+    }
+
     FakePeer &FakePeer::initialize() {
       BOOST_VERIFY_MSG(not initialized_, "Already initialized!");
       // here comes the initialization of members requiring shared_from_this()
@@ -381,7 +387,7 @@ namespace integration_framework {
 
     boost::optional<std::shared_ptr<const shared_model::interface::Proposal>>
     FakePeer::sendProposalRequest(iroha::consensus::Round round,
-                                  std::chrono::milliseconds timeout) {
+                                  std::chrono::milliseconds timeout) const {
       auto on_demand_os_transport =
           iroha::ordering::transport::OnDemandOsClientGrpcFactory(
               async_call_,

test/framework/integration_framework/fake_peer/fake_peer.hpp

@@ -68,6 +68,8 @@ namespace integration_framework {
           std::shared_ptr<iroha::ametsuchi::TxPresenceCache> tx_presence_cache,
           logger::LoggerManagerTreePtr log_manager);
 
+      ~FakePeer();
+
       /// Initialization method.
       /// \attention Must be called prior to any other instance method (except
       /// for constructor).
@@ -191,7 +193,7 @@ namespace integration_framework {
        */
       boost::optional<std::shared_ptr<const shared_model::interface::Proposal>>
       sendProposalRequest(iroha::consensus::Round round,
-                          std::chrono::milliseconds timeout);
+                          std::chrono::milliseconds timeout) const;
 
      private:
       using MstTransport = iroha::network::MstTransportGrpc;

test/framework/integration_framework/fake_peer/proposal_storage.cpp

@@ -5,6 +5,7 @@
 
 #include "framework/integration_framework/fake_peer/proposal_storage.hpp"
 
+#include <atomic>
 #include <mutex>
 
 #include <boost/range/adaptor/indirected.hpp>
@@ -18,10 +19,11 @@ namespace integration_framework {
   namespace fake_peer {
 
     ProposalStorage::ProposalStorage()
-        : default_provider_([](auto &) { return boost::none; }),
-          proposal_factory_(
+        : proposal_factory_(
               std::make_unique<shared_model::proto::ProtoProposalFactory<
-                  shared_model::validation::DefaultProposalValidator>>()) {}
+                  shared_model::validation::DefaultProposalValidator>>()) {
+      setDefaultProvider([](auto &) { return boost::none; });
+    }
 
     OrderingProposalRequestResult ProposalStorage::getProposal(
         const Round &round) {
@@ -46,7 +48,7 @@ namespace integration_framework {
       }
 
       // finally, use the defualt
-      return default_provider_(round);
+      return getDefaultProposal(round);
     }
 
     ProposalStorage &ProposalStorage::storeProposal(
@@ -114,5 +116,23 @@ namespace integration_framework {
 
     }
 
+    ProposalStorage &ProposalStorage::setDefaultProvider(DefaultProvider provider) {
+      std::atomic_store_explicit(
+          &default_provider_,
+          std::make_shared<DefaultProvider>(std::move(provider)),
+          std::memory_order_release);
+      return *this;
+    }
+
+    OrderingProposalRequestResult ProposalStorage::getDefaultProposal(
+        const Round &round) const {
+      auto default_provider = std::atomic_load_explicit(
+          &default_provider_, std::memory_order_acquire);
+      if (default_provider) {
+        return default_provider->operator()(round);
+      }
+      return {};
+    }
+
   }  // namespace fake_peer
 }  // namespace integration_framework

test/framework/integration_framework/fake_peer/proposal_storage.hpp

@@ -66,15 +66,20 @@ namespace integration_framework {
       /// with the transactions from internal storage will be returned
       void addBatches(const BatchesCollection &batches);
 
+      ProposalStorage &setDefaultProvider(DefaultProvider provider);
+
      private:
       /// Create a proposal from pending transactions, if any.
       boost::optional<std::unique_ptr<Proposal>> makeProposalFromPendingTxs(
           shared_model::interface::types::HeightType height);
 
-      DefaultProvider default_provider_;
+      OrderingProposalRequestResult getDefaultProposal(
+          const Round &round) const;
+
       std::unique_ptr<shared_model::interface::UnsafeProposalFactory>
           proposal_factory_;
 
+      std::shared_ptr<DefaultProvider> default_provider_;
       std::map<Round, std::shared_ptr<const Proposal>> proposals_map_;
       mutable std::shared_timed_mutex proposals_map_mutex_;
 

test/framework/integration_framework/integration_test_framework.cpp

@@ -401,6 +401,13 @@ namespace integration_framework {
     return iroha_instance_->getIrohaInstance()->getConsensusGate()->onOutcome();
   }
 
+  rxcpp::observable<iroha::synchronizer::SynchronizationEvent>
+  IntegrationTestFramework::getPcsOnCommitObservable() {
+    return iroha_instance_->getIrohaInstance()
+        ->getPeerCommunicationService()
+        ->on_commit();
+  }
+
   IntegrationTestFramework &IntegrationTestFramework::getTxStatus(
       const shared_model::crypto::Hash &hash,
       std::function<void(const shared_model::proto::TransactionResponse &)>

test/framework/integration_framework/integration_test_framework.hpp

@@ -386,6 +386,9 @@ namespace integration_framework {
     rxcpp::observable<iroha::network::ConsensusGate::GateObject>
     getYacOnCommitObservable();
 
+    rxcpp::observable<iroha::synchronizer::SynchronizationEvent>
+    getPcsOnCommitObservable();
+
     /**
      * Request next status of the transaction
      * @param tx_hash is hash for filtering responses

test/framework/integration_framework/iroha_instance.cpp

@@ -16,6 +16,8 @@
 
 using namespace std::chrono_literals;
 
+static constexpr std::chrono::milliseconds kMstEmissionPeriod = 100ms;
+
 namespace integration_framework {
 
   IrohaInstance::IrohaInstance(bool mst_support,
@@ -39,7 +41,12 @@ namespace integration_framework {
         // amount of minutes in a day
         mst_expiration_time_(std::chrono::minutes(24 * 60)),
         opt_mst_gossip_params_(boost::make_optional(
-            mst_support, iroha::GossipPropagationStrategyParams{})),
+            mst_support,
+            [] {
+              iroha::GossipPropagationStrategyParams params;
+              params.emission_period = kMstEmissionPeriod;
+              return params;
+            }())),
         max_rounds_delay_(0ms),
         stale_stream_max_rounds_(2),
         irohad_log_manager_(std::move(irohad_log_manager)),