Feature: request chunk protocol v2

.github/workflows/test.yml

@@ -329,6 +329,8 @@ jobs:
             test: "test-polkadot-functional-0010-validator-disabling"
           - name: "Test we are producing blocks at 6 seconds clip"
             test: "test-polkadot-functional-0011-async-backing-6-seconds-rate"
+          - name: "Systematic chunk recovery is used if the chunk mapping feature is enabled."
+            test: "test-polkadot-functional-0013-systematic-chunk-recovery"
     steps:
       - name: Download artifact
         uses: actions/download-artifact@v4

.github/workflows/zombie-tests.yaml

@@ -131,6 +131,8 @@ jobs:
             test: "test-polkadot-functional-0010-validator-disabling"
           - name: "Test we are producing blocks at 6 seconds clip"
             test: "test-polkadot-functional-0011-async-backing-6-seconds-rate"
+          - name: "Systematic chunk recovery is used if the chunk mapping feature is enabled."
+            test: "test-polkadot-functional-0013-systematic-chunk-recovery"
     steps:
       - name: Download artifact
         uses: actions/download-artifact@v4

core/injector/application_injector.cpp

@@ -123,6 +123,7 @@
 #include "network/impl/protocols/parachain_protocols.hpp"
 #include "network/impl/protocols/protocol_fetch_available_data.hpp"
 #include "network/impl/protocols/protocol_fetch_chunk.hpp"
+#include "network/impl/protocols/protocol_fetch_chunk_obsolete.hpp"
 #include "network/impl/protocols/protocol_req_collation.hpp"
 #include "network/impl/protocols/protocol_req_pov.hpp"
 #include "network/impl/protocols/send_dispute_protocol.hpp"
@@ -629,7 +630,7 @@ namespace {
                   injector.template create<sptr<application::ChainSpec>>();
               return crypto::KeyStore::Config{config.keystorePath(chain_spec->id())};
             }),
-            
+
             // inherit host injector
             libp2p::injector::makeHostInjector(
                 libp2p::injector::useWssPem(config->nodeWssPem()),

core/network/common.hpp

@@ -37,7 +37,9 @@ namespace kagome::network {
   const libp2p::peer::ProtocolName kReqCollationVStagingProtocol{
       "/{}/req_collation/2"};
   const libp2p::peer::ProtocolName kReqPovProtocol{"/{}/req_pov/1"};
-  const libp2p::peer::ProtocolName kFetchChunkProtocol{"/{}/req_chunk/1"};
+  const libp2p::peer::ProtocolName kFetchChunkProtocolObsolete{
+      "/{}/req_chunk/1"};
+  const libp2p::peer::ProtocolName kFetchChunkProtocol{"/{}/req_chunk/2"};
   const libp2p::peer::ProtocolName kFetchAvailableDataProtocol{
       "/{}/req_available_data/1"};
   const libp2p::peer::ProtocolName kFetchStatementProtocol{

core/network/helpers/stream_read_buffer.hpp

@@ -97,7 +97,7 @@ namespace kagome::network {
     const std::thread::id this_id_{std::this_thread::get_id()};
 
     void check() const {
-      BOOST_ASSERT(this_id_ == std::this_thread::get_id());
+      // BOOST_ASSERT(this_id_ == std::this_thread::get_id());
     }
 
     StreamWrapper(std::shared_ptr<libp2p::connection::StreamReadBuffer> stream)

core/network/impl/protocols/protocol_fetch_chunk.hpp

@@ -36,7 +36,7 @@ namespace kagome::network {
     ~FetchChunkProtocol() override = default;
 
     FetchChunkProtocol(libp2p::Host &host,
-                       const application::ChainSpec &chain_spec,
+                       const application::ChainSpec & /*chain_spec*/,
                        const blockchain::GenesisBlockHash &genesis_hash,
                        std::shared_ptr<parachain::ParachainProcessorImpl> pp)
         : RequestResponseProtocol<
@@ -82,7 +82,7 @@ namespace kagome::network {
                              request.chunk_index);
     }
 
-    inline static const auto kFetchChunkProtocolName = "FetchChunkProtocol"s;
+    inline static const auto kFetchChunkProtocolName = "FetchChunkProtocol_v2"s;
     std::shared_ptr<parachain::ParachainProcessorImpl> pp_;
   };
 

core/network/impl/protocols/protocol_fetch_chunk_obsolete.hpp

@@ -0,0 +1,92 @@
+/**
+ * Copyright Quadrivium LLC
+ * All Rights Reserved
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "network/protocol_base.hpp"
+
+#include <memory>
+
+#include <libp2p/connection/stream.hpp>
+#include <libp2p/host/host.hpp>
+
+#include "blockchain/genesis_block_hash.hpp"
+#include "log/logger.hpp"
+#include "network/common.hpp"
+#include "network/impl/protocols/request_response_protocol.hpp"
+#include "network/impl/stream_engine.hpp"
+#include "parachain/validator/parachain_processor.hpp"
+#include "utils/non_copyable.hpp"
+
+namespace kagome::network {
+
+  struct ReqPovProtocolImpl;
+
+  class FetchChunkProtocolObsolete final
+      : public RequestResponseProtocol<FetchChunkRequest,
+                                       FetchChunkResponseObsolete,
+                                       ScaleMessageReadWriter>,
+        NonCopyable,
+        NonMovable {
+   public:
+    FetchChunkProtocolObsolete() = delete;
+    ~FetchChunkProtocolObsolete() override = default;
+
+    FetchChunkProtocolObsolete(
+        libp2p::Host &host,
+        const application::ChainSpec & /*chain_spec*/,
+        const blockchain::GenesisBlockHash &genesis_hash,
+        std::shared_ptr<parachain::ParachainProcessorImpl> pp)
+        : RequestResponseProtocol<
+            FetchChunkRequest,
+            FetchChunkResponseObsolete,
+            ScaleMessageReadWriter>{kFetchChunkProtocolName,
+                                    host,
+                                    make_protocols(kFetchChunkProtocolObsolete,
+                                                   genesis_hash,
+                                                   kProtocolPrefixPolkadot),
+                                    log::createLogger(kFetchChunkProtocolName,
+                                                      "req_chunk_protocol")},
+          pp_{std::move(pp)} {
+      BOOST_ASSERT(pp_);
+    }
+
+   private:
+    std::optional<outcome::result<ResponseType>> onRxRequest(
+        RequestType request, std::shared_ptr<Stream> /*stream*/) override {
+      base().logger()->info("Fetching chunk request.(chunk={}, candidate={})",
+                            request.chunk_index,
+                            request.candidate);
+      auto res = pp_->OnFetchChunkRequest(std::move(request));
+      if (res.has_error()) {
+        base().logger()->error("Fetching chunk response failed.(error={})",
+                               res.error());
+        return res.as_failure();
+      }
+
+      if (auto chunk = if_type<const network::Chunk>(res.value())) {
+        base().logger()->info("Fetching chunk response with data.");
+        return outcome::success(network::ChunkObsolete{
+            .data = std::move(chunk.value().get().data),
+            .proof = std::move(chunk.value().get().proof),
+        });
+      }
+
+      base().logger()->info("Fetching chunk response empty.");
+      return outcome::success(network::Empty{});
+    }
+
+    void onTxRequest(const RequestType &request) override {
+      base().logger()->debug("Fetching chunk candidate: {}, index: {}",
+                             request.candidate,
+                             request.chunk_index);
+    }
+
+    inline static const auto kFetchChunkProtocolName = "FetchChunkProtocol_v1"s;
+    std::shared_ptr<parachain::ParachainProcessorImpl> pp_;
+  };
+
+}  // namespace kagome::network

core/network/impl/router_libp2p.cpp

@@ -18,6 +18,7 @@
 #include "network/impl/protocols/propagate_transactions_protocol.hpp"
 #include "network/impl/protocols/protocol_fetch_available_data.hpp"
 #include "network/impl/protocols/protocol_fetch_chunk.hpp"
+#include "network/impl/protocols/protocol_fetch_chunk_obsolete.hpp"
 #include "network/impl/protocols/protocol_req_collation.hpp"
 #include "network/impl/protocols/protocol_req_pov.hpp"
 #include "network/impl/protocols/send_dispute_protocol.hpp"
@@ -51,6 +52,7 @@ namespace kagome::network {
       LazySPtr<ReqCollationProtocol> req_collation_protocol,
       LazySPtr<ReqPovProtocol> req_pov_protocol,
       LazySPtr<FetchChunkProtocol> fetch_chunk_protocol,
+      LazySPtr<FetchChunkProtocolObsolete> fetch_chunk_protocol_obsolete,
       LazySPtr<FetchAvailableDataProtocol> fetch_available_data_protocol,
       LazySPtr<StatementFetchingProtocol> statement_fetching_protocol,
       LazySPtr<SendDisputeProtocol> send_dispute_protocol,
@@ -79,6 +81,8 @@ namespace kagome::network {
         req_collation_protocol_(std::move(req_collation_protocol)),
         req_pov_protocol_(std::move(req_pov_protocol)),
         fetch_chunk_protocol_(std::move(fetch_chunk_protocol)),
+        fetch_chunk_protocol_obsolete_(
+            std::move(fetch_chunk_protocol_obsolete)),
         fetch_available_data_protocol_(
             std::move(fetch_available_data_protocol)),
         statement_fetching_protocol_(std::move(statement_fetching_protocol)),
@@ -133,6 +137,7 @@ namespace kagome::network {
     lazyStart(req_collation_protocol_);
     lazyStart(req_pov_protocol_);
     lazyStart(fetch_chunk_protocol_);
+    lazyStart(fetch_chunk_protocol_obsolete_);
     lazyStart(fetch_available_data_protocol_);
     lazyStart(statement_fetching_protocol_);
     lazyStart(send_dispute_protocol_);
@@ -268,6 +273,11 @@ namespace kagome::network {
     return fetch_chunk_protocol_.get();
   }
 
+  std::shared_ptr<FetchChunkProtocolObsolete>
+  RouterLibp2p::getFetchChunkProtocolObsolete() const {
+    return fetch_chunk_protocol_obsolete_.get();
+  }
+
   std::shared_ptr<FetchAttestedCandidateProtocol>
   RouterLibp2p::getFetchAttestedCandidateProtocol() const {
     return fetch_attested_candidate_.get();

core/network/impl/router_libp2p.hpp

@@ -72,6 +72,7 @@ namespace kagome::network {
         LazySPtr<ReqCollationProtocol> req_collation_protocol,
         LazySPtr<ReqPovProtocol> req_pov_protocol,
         LazySPtr<FetchChunkProtocol> fetch_chunk_protocol,
+        LazySPtr<FetchChunkProtocolObsolete> fetch_chunk_protocol_obsolete,
         LazySPtr<FetchAvailableDataProtocol> fetch_available_data_protocol,
         LazySPtr<StatementFetchingProtocol> statement_fetching_protocol,
         LazySPtr<SendDisputeProtocol> send_dispute_protocol,
@@ -105,6 +106,8 @@ namespace kagome::network {
         const override;
     std::shared_ptr<ReqPovProtocol> getReqPovProtocol() const override;
     std::shared_ptr<FetchChunkProtocol> getFetchChunkProtocol() const override;
+    std::shared_ptr<FetchChunkProtocolObsolete> getFetchChunkProtocolObsolete()
+        const override;
     std::shared_ptr<FetchAttestedCandidateProtocol>
     getFetchAttestedCandidateProtocol() const override;
     std::shared_ptr<FetchAvailableDataProtocol> getFetchAvailableDataProtocol()
@@ -155,6 +158,7 @@ namespace kagome::network {
     LazySPtr<ReqCollationProtocol> req_collation_protocol_;
     LazySPtr<ReqPovProtocol> req_pov_protocol_;
     LazySPtr<FetchChunkProtocol> fetch_chunk_protocol_;
+    LazySPtr<FetchChunkProtocolObsolete> fetch_chunk_protocol_obsolete_;
     LazySPtr<FetchAvailableDataProtocol> fetch_available_data_protocol_;
     LazySPtr<StatementFetchingProtocol> statement_fetching_protocol_;
 

core/network/peer_state.hpp

@@ -62,7 +62,8 @@ namespace kagome::network {
     std::optional<CollatingPeerState> collator_state = std::nullopt;
     View view;
     std::unordered_set<common::Hash256> implicit_view;
-    std::optional<CollationVersion> version;
+    std::optional<CollationVersion> collation_version;
+    std::optional<ReqChunkVersion> req_chunk_version;
 
     /// Update the view, returning a vector of implicit relay-parents which
     /// weren't previously part of the view.

core/network/router.hpp

@@ -19,6 +19,7 @@ namespace kagome::network {
   class ReqCollationProtocol;
   class ReqPovProtocol;
   class FetchChunkProtocol;
+  class FetchChunkProtocolObsolete;
   class FetchAvailableDataProtocol;
   class StatementFetchingProtocol;
   class PropagateTransactionsProtocol;
@@ -56,6 +57,8 @@ namespace kagome::network {
     virtual std::shared_ptr<ReqPovProtocol> getReqPovProtocol() const = 0;
     virtual std::shared_ptr<FetchChunkProtocol> getFetchChunkProtocol()
         const = 0;
+    virtual std::shared_ptr<FetchChunkProtocolObsolete>
+    getFetchChunkProtocolObsolete() const = 0;
     virtual std::shared_ptr<FetchAttestedCandidateProtocol>
     getFetchAttestedCandidateProtocol() const = 0;
     virtual std::shared_ptr<FetchAvailableDataProtocol>

core/network/types/collator_messages.hpp

@@ -68,7 +68,7 @@ namespace kagome::network {
     /// The erasure-encoded chunk of data belonging to the candidate block.
     common::Buffer chunk;
     /// The index of this erasure-encoded chunk of data.
-    ValidatorIndex index;
+    ChunkIndex index;
     /// Proof for this chunk's branch in the Merkle tree.
     ChunkProof proof;
   };
@@ -176,19 +176,38 @@ namespace kagome::network {
     SCALE_TIE(2);
 
     CandidateHash candidate;  /// parachain candidate hash
-    uint32_t chunk_index;     /// index of the chunk
+    ChunkIndex chunk_index;   /// index of the chunk
   };
 
   /**
    * Sent by nodes to the clients who issued a chunk fetching request.
+   * Version 1 (obsolete)
    */
-  struct Chunk {
+  struct ChunkObsolete {
     SCALE_TIE(2);
 
-    common::Buffer data;  /// chunk data
-    ChunkProof proof;     /// chunk proof
+    /// chunk data
+    common::Buffer data;
+    /// chunk proof
+    ChunkProof proof;
+  };
+  using FetchChunkResponseObsolete = boost::variant<ChunkObsolete, Empty>;
+
+  /**
+   * Sent by nodes to the clients who issued a chunk fetching request. Version 2
+   */
+  struct Chunk {
+    SCALE_TIE(3);
+
+    /// chunk data
+    common::Buffer data;
+    /// chunk index
+    ChunkIndex chunk_index;
+    /// chunk proof
+    ChunkProof proof;
   };
   using FetchChunkResponse = boost::variant<Chunk, Empty>;
+
   using FetchAvailableDataRequest = CandidateHash;
   using FetchAvailableDataResponse =
       boost::variant<runtime::AvailableData, Empty>;

core/network/types/collator_messages_vstaging.hpp

@@ -353,13 +353,20 @@ namespace kagome::network::vstaging {
 
 namespace kagome::network {
 
-  enum CollationVersion {
+  enum class CollationVersion {
     /// The first version.
     V1 = 1,
     /// The staging version.
     VStaging = 2,
   };
 
+  enum class ReqChunkVersion {
+    /// The first (obsolete) version.
+    V1_obsolete = 1,
+    /// The second version.
+    V2 = 2,
+  };
+
   /// Candidate supplied with a para head it's built on top of.
   /// polkadot/node/network/collator-protocol/src/validator_side/collation.rs
   struct ProspectiveCandidate {