Fix/kusama grid size fixes

core/authority_discovery/query/query_impl.cpp

@@ -108,7 +108,9 @@ namespace kagome::authority_discovery {
     }
     auto r = add(*id, value);
     if (not r) {
-      SL_DEBUG(log_, "Can't add: {}", r.error());
+      SL_WARN(log_, "Can't add: {}", r.error());
+    } else {
+      SL_INFO(log_, "Added or updated success: {}", *id);
     }
     return r;
   }

core/network/impl/protocols/protocol_fetch_available_data.hpp

@@ -49,6 +49,9 @@ namespace kagome::network {
    private:
     std::optional<outcome::result<ResponseType>> onRxRequest(
         RequestType candidate_hash, std::shared_ptr<Stream>) override {
+      base().logger()->trace("Fetch available data .(candidate hash={})",
+                             candidate_hash);
+
       if (auto r = av_store_->getPovAndData(candidate_hash)) {
         return std::move(*r);
       }

core/parachain/approval/approval.hpp

@@ -208,6 +208,9 @@ namespace kagome::parachain::approval {
   inline bool is_local_approval(const ApprovalStateTransition &val) {
     return boost::get<LocalApproval>(&val) != nullptr;
   }
+  inline bool is_remote_approval(const ApprovalStateTransition &val) {
+    return boost::get<RemoteApproval>(&val) != nullptr;
+  }
 
   /// Metadata about a block which is now live in the approval protocol.
   struct BlockApprovalMeta {

core/parachain/approval/approval_distribution.cpp

@@ -578,7 +578,7 @@ namespace kagome::parachain {
       common::MainThreadPool &main_thread_pool,
       LazySPtr<dispute::DisputeCoordinator> dispute_coordinator)
       : approval_thread_handler_{poolHandlerReadyMake(
-          this, app_state_manager, approval_thread_pool, logger_)},
+            this, app_state_manager, approval_thread_pool, logger_)},
         worker_pool_handler_{worker_thread_pool.handler(*app_state_manager)},
         parachain_host_(std::move(parachain_host)),
         slots_util_(slots_util),
@@ -1469,25 +1469,55 @@ namespace kagome::parachain {
       blocks_by_number_[meta.number].insert(meta.hash);
     }
 
+    std::unordered_set<libp2p::peer::PeerId> active_peers;
+    pm_->enumeratePeerState(
+        [&](const libp2p::peer::PeerId &peer, network::PeerState &_) {
+          active_peers.insert(peer);
+          return true;
+        });
+
+    network::View our_current_view{
+        .heads_ = block_tree_->getLeaves(),
+        .finalized_number_ = block_tree_->getLastFinalized().number,
+    };
+
     approval_thread_handler_->execute([wself{weak_from_this()},
+                                       our_current_view{
+                                           std::move(our_current_view)},
+                                       active_peers{std::move(active_peers)},
                                        new_hash,
-                                       finalized_block_number,
                                        meta{std::move(meta)}]() {
       if (auto self = wself.lock()) {
         SL_TRACE(self->logger_, "Got new block.(hash={})", new_hash);
+        // std::unordered_map<libp2p::peer::PeerId, network::View>
+        // peer_views(std::move(self->peer_views_));
+        for (auto it = self->peer_views_.begin();
+             it != self->peer_views_.end();) {
+          if (active_peers.contains(it->first)) {
+            ++it;
+          } else {
+            it = self->peer_views_.erase(it);
+          }
+        }
+
+        for (const auto &p : active_peers) {
+          std::ignore = self->peer_views_[p];
+        }
+
         for (const auto &[peed_id, view] : self->peer_views_) {
-          for (const auto &h : view.heads_) {
-            if (h == meta.hash) {
-              self->unify_with_peer(
-                  self->storedDistribBlockEntries(),
-                  peed_id,
-                  network::View{
-                      .heads_ = {h},
-                      .finalized_number_ = finalized_block_number,
-                  },
-                  false);
-            }
+          network::View view_intersection{
+              .heads_ = {},
+              .finalized_number_ = view.finalized_number_,
+          };
+
+          if (new_hash && view.contains(*new_hash)) {
+            view_intersection.heads_.emplace_back(*new_hash);
           }
+
+          self->unify_with_peer(self->storedDistribBlockEntries(),
+                                peed_id,
+                                view_intersection,
+                                false);
         }
 
         for (auto it = self->pending_known_.begin();
@@ -2134,8 +2164,10 @@ namespace kagome::parachain {
             ApprovalRouting{
                 .required_routing =
                     grid::RequiredRouting{
-                        .value = grid::RequiredRouting::
-                            GridXY},  /// TODO(iceseer): calculate based on grid
+                        .value =
+                            grid::RequiredRouting::All},  /// TODO(iceseer):
+                                                          /// calculate
+                                                          /// based on grid
                 .local = local,
                 .random_routing = {},
                 .peers_randomly_routed = {},
@@ -2547,17 +2579,36 @@ namespace kagome::parachain {
     auto se = pm_->getStreamEngine();
     BOOST_ASSERT(se);
 
-    se->broadcast(
-        router_->getValidationProtocolVStaging(),
-        std::make_shared<
+    auto msg = std::make_shared<
             network::WireMessage<network::vstaging::ValidatorProtocolMessage>>(
             network::vstaging::ApprovalDistributionMessage{
                 network::vstaging::Assignments{
                     .assignments = {network::vstaging::Assignment{
                         .indirect_assignment_cert = indirect_cert,
                         .candidate_bitfield = candidate_indices,
-                    }}}}),
-        [&](const libp2p::peer::PeerId &p) { return peers.contains(p); });
+                    }}}});
+
+    for (const auto &peer : peers) {
+      parachain_processor_->tryOpenOutgoingValidationStream(
+          peer,
+          network::CollationVersion::VStaging,
+          [WEAK_SELF, peer{peer}, se, msg](auto &&stream) {
+            WEAK_LOCK(self);
+            se->send(peer, self->router_->getValidationProtocolVStaging(), msg);
+          });
+    }
+
+//    se->broadcast(
+//        router_->getValidationProtocolVStaging(),
+//        std::make_shared<
+//            network::WireMessage<network::vstaging::ValidatorProtocolMessage>>(
+//            network::vstaging::ApprovalDistributionMessage{
+//                network::vstaging::Assignments{
+//                    .assignments = {network::vstaging::Assignment{
+//                        .indirect_assignment_cert = indirect_cert,
+//                        .candidate_bitfield = candidate_indices,
+//                    }}}}),
+//        [&](const libp2p::peer::PeerId &p) { return peers.count(p) != 0ull; });
   }
 
   void ApprovalDistribution::send_assignments_batched(
@@ -2668,15 +2719,22 @@ namespace kagome::parachain {
     auto se = pm_->getStreamEngine();
     BOOST_ASSERT(se);
 
-    se->broadcast(
-        router_->getValidationProtocolVStaging(),
-        std::make_shared<
+    auto msg = std::make_shared<
             network::WireMessage<network::vstaging::ValidatorProtocolMessage>>(
             network::vstaging::ApprovalDistributionMessage{
                 network::vstaging::Approvals{
                     .approvals = {vote},
-                }}),
-        [&](const libp2p::peer::PeerId &p) { return peers.contains(p); });
+                }});
+
+    for (const auto &peer : peers) {
+      parachain_processor_->tryOpenOutgoingValidationStream(
+          peer,
+          network::CollationVersion::VStaging,
+          [WEAK_SELF, peer{peer}, se, msg](auto &&stream) {
+            WEAK_LOCK(self);
+            se->send(peer, self->router_->getValidationProtocolVStaging(), msg);
+          });
+    }
   }
 
   void ApprovalDistribution::issue_approval(const CandidateHash &candidate_hash,
@@ -2887,9 +2945,10 @@ namespace kagome::parachain {
             };
             return approval::min_or_some(
                 e.next_no_show,
-                (e.last_assignment_tick ? filter(
-                     *e.last_assignment_tick + kApprovalDelay, tick_now)
-                                        : std::optional<Tick>{}));
+                (e.last_assignment_tick
+                     ? filter(*e.last_assignment_tick + kApprovalDelay,
+                              tick_now)
+                     : std::optional<Tick>{}));
           },
           [&](const approval::PendingRequiredTranche &e) {
             std::optional<DelayTranche> next_announced{};
@@ -3036,6 +3095,37 @@ namespace kagome::parachain {
                            status.block_tick,
                            tick_now,
                            status.required_tranches);
+
+    if (is_approved && is_remote_approval(transition)) {
+      for (const auto &[fork_block_hash, fork_approval_entry] :
+           candidate_entry.block_assignments) {
+        if (fork_block_hash == block_hash) {
+          continue;
+        }
+
+        bool assigned_on_fork_block = false;
+        if (validator_index) {
+          assigned_on_fork_block =
+              fork_approval_entry.is_assigned(*validator_index);
+        }
+
+        if (!wakeup_for(fork_block_hash, candidate_hash)
+            && !fork_approval_entry.approved && assigned_on_fork_block) {
+          auto opt_fork_block_entry = storedBlockEntries().get(fork_block_hash);
+          if (!opt_fork_block_entry) {
+            SL_TRACE(logger_,
+                     "Failed to load block entry. (fork_block_hash={})",
+                     fork_block_hash);
+          } else {
+            runScheduleWakeup(fork_block_hash,
+                              opt_fork_block_entry->get().block_number,
+                              candidate_hash,
+                              tick_now + 1);
+          }
+        }
+      }
+    }
+
     if (approval::is_local_approval(transition) || newly_approved
         || (already_approved_by && !*already_approved_by)) {
       BOOST_ASSERT(storedCandidateEntries().get(candidate_hash)->get()
@@ -3108,6 +3198,12 @@ namespace kagome::parachain {
     target_block[candidate_hash].emplace_back(tick, std::move(handle));
   }
 
+  bool ApprovalDistribution::wakeup_for(const primitives::BlockHash &block_hash,
+                                        const CandidateHash &candidate_hash) {
+    auto it = active_tranches_.find(block_hash);
+    return it != active_tranches_.end() && it->second.contains(candidate_hash);
+  }
+
   void ApprovalDistribution::handleTranche(
       const primitives::BlockHash &block_hash,
       primitives::BlockNumber block_number,

core/parachain/approval/approval_distribution.hpp

@@ -136,7 +136,7 @@ namespace kagome::parachain {
       }
 
       /// Whether a validator is already assigned.
-      bool is_assigned(ValidatorIndex validator_index) {
+      bool is_assigned(ValidatorIndex validator_index) const {
         if (validator_index < assignments.bits.size()) {
           return assignments.bits[validator_index];
         }
@@ -233,8 +233,9 @@ namespace kagome::parachain {
       CandidateEntry(const network::CandidateReceipt &receipt,
                      SessionIndex session_index,
                      size_t approvals_size)
-          : CandidateEntry(
-              HashedCandidateReceipt{receipt}, session_index, approvals_size) {}
+          : CandidateEntry(HashedCandidateReceipt{receipt},
+                           session_index,
+                           approvals_size) {}
 
       std::optional<std::reference_wrapper<ApprovalEntry>> approval_entry(
           const network::RelayHash &relay_hash) {
@@ -792,6 +793,9 @@ namespace kagome::parachain {
     void scheduleTranche(const primitives::BlockHash &head,
                          BlockImportedCandidates &&candidate);
 
+    bool wakeup_for(const primitives::BlockHash &block_hash,
+                    const CandidateHash &candidate_hash);
+
     void runDistributeAssignment(
         const approval::IndirectAssignmentCertV2 &indirect_cert,
         const scale::BitVec &candidate_indices,

core/parachain/availability/bitfield/signer.cpp

@@ -56,20 +56,6 @@ namespace kagome::parachain {
     for (auto &candidate : candidates) {
       bitfield.bits.push_back(
           candidate && store_->hasChunk(*candidate, signer.validatorIndex()));
-      if (candidate) {
-        SL_TRACE(logger_,
-                 "Signing bitfields.(relay_parent={}, validator index={}, has "
-                 "chunk={})",
-                 relay_parent,
-                 signer.validatorIndex(),
-                 bitfield.bits.back() ? 1 : 0);
-      } else {
-        SL_TRACE(logger_,
-                 "Signing bitfields.(relay_parent={}, validator index={}, NOT "
-                 "OCCUPIED)",
-                 relay_parent,
-                 signer.validatorIndex());
-      }
     }
 
     OUTCOME_TRY(signed_bitfield, signer.sign(bitfield));

core/parachain/availability/recovery/recovery_impl.cpp

@@ -791,6 +791,11 @@ namespace kagome::parachain {
     OUTCOME_TRY(chunks, toChunks(active.chunks_total, data));
     auto root = makeTrieProof(chunks);
     if (root != active.erasure_encoding_root) {
+      SL_TRACE(logger_,
+               "Trie root mismatch. (root={}, ref root={}, n_validators={})",
+               root,
+               active.erasure_encoding_root,
+               active.validators.size());
       return ErasureCodingRootError::MISMATCH;
     }
     return outcome::success();

core/parachain/backing/grid.hpp

@@ -101,7 +101,11 @@ namespace kagome::parachain::grid {
 
   /// View for one group
   struct View {
-    std::unordered_set<ValidatorIndex> receiving, sending;
+    std::unordered_set<ValidatorIndex> sending, receiving;
+
+    bool operator==(const View &r) const {
+      return sending == r.sending && receiving == r.receiving;
+    }
 
     bool canReceive(bool full, ValidatorIndex from) const {
       return (full ? receiving : sending).contains(from);
@@ -237,12 +241,8 @@ namespace kagome::parachain::grid {
   }
 
   inline std::vector<ValidatorIndex> shuffle(
-      const std::vector<std::vector<ValidatorIndex>> &groups,
+      size_t n,
       std::span<const uint8_t, 32> babe_randomness) {
-    size_t n = 0;
-    for (auto &group : groups) {
-      n += group.size();
-    }
     std::vector<ValidatorIndex> validators;
     validators.resize(n);
     std::iota(validators.begin(), validators.end(), 0);