Fix/finalization freeze

core/dispute_coordinator/impl/dispute_coordinator_impl.cpp

@@ -1847,7 +1847,7 @@ namespace kagome::dispute {
       return cb(res.as_failure());
     }
 
-    auto &valid_import = res.value();
+    [[maybe_unused]] auto &valid_import = res.value();
 
     return cb(outcome::success());
   }

core/parachain/CMakeLists.txt

@@ -25,4 +25,5 @@ target_link_libraries(validator_parachain
     crypto_store
     network
     erasure_coding_crust::ec-cpp
+    waitable_timer
     )

core/parachain/approval/approval_distribution.hpp

@@ -28,6 +28,7 @@
 #include "network/peer_view.hpp"
 #include "network/types/collator_messages.hpp"
 #include "parachain/approval/approved_ancestor.hpp"
+#include "parachain/approval/knowledge.hpp"
 #include "parachain/approval/store.hpp"
 #include "parachain/availability/recovery/recovery.hpp"
 #include "parachain/validator/parachain_processor.hpp"
@@ -107,13 +108,13 @@ namespace kagome::parachain {
 
       ApprovalEntry(
           GroupIndex group_index,
-          std::optional<std::reference_wrapper<OurAssignment>> assignment,
+          std::optional<std::reference_wrapper<const OurAssignment>> assignment,
           size_t assignments_size)
           : backing_group{group_index},
             our_assignment{assignment},
             approved(false) {
         assignments.bits.insert(
-            assignments.bits.begin(), assignments_size, false);
+            assignments.bits.end(), assignments_size, false);
       }
 
       /// Whether a validator is already assigned.
@@ -208,7 +209,7 @@ namespace kagome::parachain {
                      SessionIndex session_index,
                      size_t approvals_size)
           : candidate(receipt), session(session_index) {
-        approvals.bits.insert(approvals.bits.begin(), approvals_size, false);
+        approvals.bits.insert(approvals.bits.end(), approvals_size, false);
       }
 
       std::optional<std::reference_wrapper<ApprovalEntry>> approval_entry(
@@ -278,6 +279,19 @@ namespace kagome::parachain {
     /// AppStateManager impl
     bool prepare();
 
+    using CandidateIncludedList =
+        std::vector<std::tuple<CandidateHash,
+                               network::CandidateReceipt,
+                               CoreIndex,
+                               GroupIndex>>;
+    using AssignmentsList = std::unordered_map<CoreIndex, OurAssignment>;
+
+    static AssignmentsList compute_assignments(
+        const std::shared_ptr<crypto::CryptoStore> &keystore,
+        const runtime::SessionInfo &config,
+        const RelayVRFStory &relay_vrf_story,
+        const CandidateIncludedList &leaving_cores);
+
     void onValidationProtocolMsg(
         const libp2p::peer::PeerId &peer_id,
         const network::ValidatorProtocolMessage &message);
@@ -296,41 +310,31 @@ namespace kagome::parachain {
         const primitives::BlockInfo &max) const override;
 
    private:
-    using CandidateIncludedList =
-        std::vector<std::tuple<CandidateHash,
-                               network::CandidateReceipt,
-                               CoreIndex,
-                               GroupIndex>>;
-    using AssignmentsList = std::unordered_map<CoreIndex, OurAssignment>;
-
     struct ImportedBlockInfo {
       CandidateIncludedList included_candidates;
       SessionIndex session_index;
       AssignmentsList assignments;
       size_t n_validators;
       RelayVRFStory relay_vrf_story;
       consensus::babe::BabeSlotNumber slot;
-      runtime::SessionInfo session_info;
       std::optional<primitives::BlockNumber> force_approve;
     };
 
     struct ApprovingContext {
       primitives::BlockHeader block_header;
       std::optional<CandidateIncludedList> included_candidates;
-      std::optional<SessionIndex> session_index;
       std::optional<consensus::babe::BabeBlockHeader> babe_block_header;
       std::optional<consensus::babe::EpochNumber> babe_epoch;
       std::optional<primitives::Randomness> randomness;
       std::optional<primitives::AuthorityList> authorities;
-      std::optional<runtime::SessionInfo> session_info;
 
       std::shared_ptr<boost::asio::io_context> complete_callback_context;
       std::function<void(outcome::result<ImportedBlockInfo> &&)>
           complete_callback;
 
       bool is_complete() const {
-        return included_candidates && babe_epoch && session_index
-            && session_info && babe_block_header && randomness && authorities;
+        return included_candidates && babe_epoch && babe_block_header
+            && randomness && authorities;
       }
     };
 
@@ -354,14 +358,24 @@ namespace kagome::parachain {
     /// for the same candidate, if it is included by multiple blocks - this is
     /// likely the case when there are forks.
     struct DistribCandidateEntry {
-      std::unordered_map<ValidatorIndex, MessageState> messages;
+      std::unordered_map<ValidatorIndex, MessageState> messages{};
     };
 
     /// Information about blocks in our current view as well as whether peers
     /// know of them.
     struct DistribBlockEntry {
       /// A votes entry for each candidate indexed by [`CandidateIndex`].
-      std::vector<DistribCandidateEntry> candidates;
+      std::vector<DistribCandidateEntry> candidates{};
+      /// Our knowledge of messages.
+      approval::Knowledge knowledge{};
+      /// Peers who we know are aware of this block and thus, the candidates
+      /// within it. This maps to their knowledge of messages.
+      std::unordered_map<libp2p::peer::PeerId, approval::PeerKnowledge>
+          known_by{};
+      /// The number of the block.
+      primitives::BlockNumber number;
+      /// The parent hash of the block.
+      RelayHash parent_hash;
     };
 
     /// Metadata regarding approval of a particular block, by way of approval of
@@ -371,7 +385,6 @@ namespace kagome::parachain {
       primitives::BlockHash parent_hash;
       primitives::BlockNumber block_number;
       SessionIndex session;
-      runtime::SessionInfo session_info;
       consensus::babe::BabeSlotNumber slot;
       RelayVRFStory relay_vrf_story;
       // The candidates included as-of this block and the index of the core they
@@ -420,11 +433,12 @@ namespace kagome::parachain {
 
     /// Information about a block and imported candidates.
     struct BlockImportedCandidates {
-      primitives::BlockHash block_hash;
-      primitives::BlockNumber block_number;
-      network::Tick block_tick;
-      network::Tick no_show_duration;
-      std::vector<std::pair<CandidateHash, CandidateEntry>> imported_candidates;
+      primitives::BlockHash block_hash{};
+      primitives::BlockNumber block_number{};
+      network::Tick block_tick{};
+      network::Tick no_show_duration{};
+      std::vector<std::pair<CandidateHash, CandidateEntry>>
+          imported_candidates{};
     };
 
     using AssignmentOrApproval =
@@ -476,12 +490,6 @@ namespace kagome::parachain {
                               primitives::AuthorityList,
                               primitives::Randomness>>;
 
-    AssignmentsList compute_assignments(
-        const std::shared_ptr<crypto::CryptoStore> &keystore,
-        const runtime::SessionInfo &config,
-        const RelayVRFStory &relay_vrf_story,
-        const CandidateIncludedList &leaving_cores);
-
     void imported_block_info(const primitives::BlockHash &block_hash,
                              const primitives::BlockHeader &block_header);
 
@@ -529,24 +537,32 @@ namespace kagome::parachain {
     template <typename Func>
     void for_ACU(const primitives::BlockHash &block_hash, Func &&func);
 
-    void try_process_approving_context(ApprovingContextUnit &acu);
+    void try_process_approving_context(
+        ApprovingContextUnit &acu,
+        SessionIndex session_index,
+        const runtime::SessionInfo &session_info);
 
-    std::optional<std::pair<ValidatorIndex, crypto::Sr25519Keypair>>
+    static std::optional<std::pair<ValidatorIndex, crypto::Sr25519Keypair>>
     findAssignmentKey(const std::shared_ptr<crypto::CryptoStore> &keystore,
                       const runtime::SessionInfo &config);
 
+    void unify_with_peer(StoreUnit<StorePair<Hash, DistribBlockEntry>> &entries,
+                         const libp2p::peer::PeerId &peer_id,
+                         const network::View &view);
+
     outcome::result<BlockImportedCandidates> processImportedBlock(
         primitives::BlockNumber block_number,
         const primitives::BlockHash &block_hash,
         const primitives::BlockHash &parent_hash,
+        primitives::BlockNumber finalized_block_number,
         ImportedBlockInfo &&block_info);
 
     outcome::result<std::vector<std::pair<CandidateHash, CandidateEntry>>>
     add_block_entry(primitives::BlockNumber block_number,
                     const primitives::BlockHash &block_hash,
                     const primitives::BlockHash &parent_hash,
                     scale::BitVec &&approved_bitfield,
-                    ImportedBlockInfo &&block_info);
+                    const ImportedBlockInfo &block_info);
 
     void on_active_leaves_update(const network::ExView &updated);
 
@@ -576,7 +592,8 @@ namespace kagome::parachain {
         const network::CandidateReceipt &candidate,
         GroupIndex backing_group);
 
-    void runNewBlocks(approval::BlockApprovalMeta &&approval_meta);
+    void runNewBlocks(approval::BlockApprovalMeta &&approval_meta,
+                      primitives::BlockNumber finalized_block_number);
 
     std::optional<ValidatorSignature> sign_approval(
         const crypto::Sr25519PublicKey &pubkey,
@@ -607,9 +624,19 @@ namespace kagome::parachain {
 
     void runDistributeAssignment(
         const approval::IndirectAssignmentCert &indirect_cert,
-        CandidateIndex candidate_index);
+        CandidateIndex candidate_index,
+        std::unordered_set<libp2p::peer::PeerId> &&peers);
+
+    void send_assignments_batched(std::deque<network::Assignment> &&assignments,
+                                  const libp2p::peer::PeerId &peer_id);
 
-    void runDistributeApproval(const network::IndirectSignedApprovalVote &vote);
+    void send_approvals_batched(
+        std::deque<network::IndirectSignedApprovalVote> &&approvals,
+        const libp2p::peer::PeerId &peer_id);
+
+    void runDistributeApproval(
+        const network::IndirectSignedApprovalVote &vote,
+        std::unordered_set<libp2p::peer::PeerId> &&peers);
 
     void runScheduleWakeup(const primitives::BlockHash &block_hash,
                            primitives::BlockNumber block_number,
@@ -618,6 +645,9 @@ namespace kagome::parachain {
 
     void clearCaches(const primitives::events::ChainEventParams &event);
 
+    void store_remote_view(const libp2p::peer::PeerId &peer_id,
+                           const network::View &view);
+
     auto &storedBlocks() {
       return as<StorePair<primitives::BlockNumber,
                           std::unordered_set<network::Hash>>>(store_);
@@ -653,6 +683,7 @@ namespace kagome::parachain {
     const ApprovalVotingSubsystem config_;
     std::shared_ptr<network::PeerView> peer_view_;
     network::PeerView::MyViewSubscriberPtr my_view_sub_;
+    network::PeerView::PeerViewSubscriberPtr remote_view_sub_;
     std::shared_ptr<primitives::events::ChainEventSubscriber> chain_sub_;
 
     Store<StorePair<primitives::BlockNumber, std::unordered_set<Hash>>,
@@ -676,14 +707,24 @@ namespace kagome::parachain {
         Hash,
         std::vector<std::pair<libp2p::peer::PeerId, PendingMessage>>>
         pending_known_;
+    std::unordered_map<libp2p::peer::PeerId, network::View> peer_views_;
+    std::map<primitives::BlockNumber, std::unordered_set<primitives::BlockHash>>
+        blocks_by_number_;
 
     /// thread_pool_ context access
-    using ScheduledCandidateTimer =
-        std::unordered_map<CandidateHash,
-                           std::pair<Tick, std::unique_ptr<clock::Timer>>>;
+    using ScheduledCandidateTimer = std::unordered_map<
+        CandidateHash,
+        std::vector<std::pair<Tick, std::unique_ptr<clock::Timer>>>>;
     std::unordered_map<network::BlockHash, ScheduledCandidateTimer>
         active_tranches_;
 
+    struct ApprovalCache {
+      std::unordered_set<primitives::BlockHash> blocks_;
+      ApprovalOutcome approval_result;
+    };
+    SafeObject<std::unordered_map<CandidateHash, ApprovalCache>, std::mutex>
+        approvals_cache_;
+
     log::Logger logger_ =
         log::createLogger("ApprovalDistribution", "parachain");
   };

core/parachain/approval/knowledge.hpp

@@ -0,0 +1,80 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef KAGOME_KNOWLEDGE_HPP
+#define KAGOME_KNOWLEDGE_HPP
+
+#include <tuple>
+
+#include "common/visitor.hpp"
+#include "consensus/babe/common.hpp"
+#include "consensus/validation/prepare_transcript.hpp"
+#include "outcome/outcome.hpp"
+#include "parachain/approval/state.hpp"
+#include "parachain/types.hpp"
+
+namespace kagome::parachain::approval {
+
+  enum struct MessageKind { Assignment, Approval };
+  using MessageSubject = std::tuple<Hash, CandidateIndex, ValidatorIndex>;
+
+  struct MessageSubjectHash {
+    auto operator()(const MessageSubject &obj) const {
+      size_t value{0ull};
+      std::apply(
+          [&](const auto &...v) { (..., boost::hash_combine(value, v)); }, obj);
+      return value;
+    }
+  };
+
+  struct Knowledge {
+    // When there is no entry, this means the message is unknown
+    // When there is an entry with `MessageKind::Assignment`, the assignment is
+    // known. When there is an entry with `MessageKind::Approval`, the
+    // assignment and approval are known.
+    std::unordered_map<MessageSubject, MessageKind, MessageSubjectHash>
+        known_messages{};
+
+    bool contains(const MessageSubject &message,
+                  const MessageKind &kind) const {
+      auto it = known_messages.find(message);
+      if (it == known_messages.end()) {
+        return false;
+      }
+      if (MessageKind::Assignment == kind) {
+        return true;
+      }
+      return MessageKind::Approval == it->second;
+    }
+
+    bool insert(const MessageSubject &message, const MessageKind &kind) {
+      const auto &[it, inserted] = known_messages.emplace(message, kind);
+      if (inserted) {
+        return true;
+      }
+      if (MessageKind::Assignment == it->second
+          && MessageKind::Approval == kind) {
+        it->second = MessageKind::Approval;
+        return true;
+      }
+      return false;
+    }
+  };
+
+  struct PeerKnowledge {
+    /// The knowledge we've sent to the peer.
+    Knowledge sent{};
+    /// The knowledge we've received from the peer.
+    Knowledge received{};
+
+    bool contains(const MessageSubject &message,
+                  const MessageKind &kind) const {
+      return sent.contains(message, kind) || received.contains(message, kind);
+    }
+  };
+
+}  // namespace kagome::parachain::approval
+
+#endif  // KAGOME_KNOWLEDGE_HPP

test/core/CMakeLists.txt

@@ -24,3 +24,4 @@ add_subdirectory(subscription)
 add_subdirectory(telemetry)
 add_subdirectory(transaction_pool)
 add_subdirectory(dispute_coordinator)
+add_subdirectory(parachain)

test/core/parachain/CMakeLists.txt

@@ -0,0 +1,13 @@
+#
+# Copyright Soramitsu Co., Ltd. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+
+addtest(assignments_test
+    assignments.cpp
+    )
+target_link_libraries(assignments_test
+    crypto_store
+    base_fs_test
+    validator_parachain
+    )