interface

Signed-off-by: iceseer <[email protected]>

core/consensus/babe/impl/babe.hpp

@@ -71,7 +71,7 @@ namespace kagome::offchain {
 
 namespace kagome::parachain {
   class BitfieldStore;
-  class ParachainProcessorImpl;
+  class ParachainProcessor;
   struct BackedCandidatesSource;
 }  // namespace kagome::parachain
 

core/injector/application_injector.cpp

@@ -803,6 +803,10 @@ namespace {
                 [](const auto &injector) {
                   return injector.template create<sptr<parachain::ParachainProcessorImpl>>();
                 }),
+            bind_by_lambda<parachain::ParachainStorage>(
+                [](const auto &injector) {
+                  return injector.template create<sptr<parachain::ParachainProcessorImpl>>();
+                }),
             bind_by_lambda<parachain::ParachainProcessor>(
                 [](const auto &injector) {
                   return injector.template create<sptr<parachain::ParachainProcessorImpl>>();
@@ -1032,7 +1036,7 @@ namespace kagome::injector {
         .template create<sptr<parachain::ParachainObserver>>();
   }
 
-  std::shared_ptr<parachain::ParachainProcessorImpl>
+  std::shared_ptr<parachain::ParachainProcessor>
   KagomeNodeInjector::injectParachainProcessor() {
     return pimpl_->injector_
         .template create<sptr<parachain::ParachainProcessorImpl>>();

core/injector/application_injector.hpp

@@ -58,7 +58,7 @@ namespace kagome {
 
   namespace parachain {
     class ParachainObserver;
-    class ParachainProcessorImpl;
+    class ParachainProcessor;
     class ApprovalDistribution;
 
     namespace statement_distribution {
@@ -134,7 +134,7 @@ namespace kagome::injector {
     std::shared_ptr<network::SyncProtocolObserver> injectSyncObserver();
     std::shared_ptr<network::StateProtocolObserver> injectStateObserver();
     std::shared_ptr<parachain::ParachainObserver> injectParachainObserver();
-    std::shared_ptr<parachain::ParachainProcessorImpl>
+    std::shared_ptr<parachain::ParachainProcessor>
     injectParachainProcessor();
     std::shared_ptr<parachain::statement_distribution::StatementDistribution>
     injectStatementDistribution();

core/network/impl/protocols/protocol_fetch_chunk.hpp

@@ -42,7 +42,7 @@ namespace kagome::network {
         libp2p::Host &host,
         const application::ChainSpec & /*chain_spec*/,
         const blockchain::GenesisBlockHash &genesis_hash,
-        std::shared_ptr<parachain::ParachainProcessorImpl> pp,
+        std::shared_ptr<parachain::ParachainStorage> pp,
         std::shared_ptr<PeerManager> pm)
         : RequestResponseProtocolImpl<
             FetchChunkRequest,
@@ -130,7 +130,7 @@ namespace kagome::network {
     }
 
     inline static const auto kFetchChunkProtocolName = "FetchChunkProtocol_v2"s;
-    std::shared_ptr<parachain::ParachainProcessorImpl> pp_;
+    std::shared_ptr<parachain::ParachainStorage> pp_;
     std::shared_ptr<PeerManager> pm_;
   };
 

core/network/impl/protocols/protocol_fetch_chunk_obsolete.hpp

@@ -42,7 +42,7 @@ namespace kagome::network {
         libp2p::Host &host,
         const application::ChainSpec & /*chain_spec*/,
         const blockchain::GenesisBlockHash &genesis_hash,
-        std::shared_ptr<parachain::ParachainProcessorImpl> pp)
+        std::shared_ptr<parachain::ParachainStorage> pp)
         : RequestResponseProtocolImpl<
             FetchChunkRequest,
             FetchChunkResponseObsolete,
@@ -114,7 +114,7 @@ namespace kagome::network {
     }
 
     inline static const auto kFetchChunkProtocolName = "FetchChunkProtocol_v1"s;
-    std::shared_ptr<parachain::ParachainProcessorImpl> pp_;
+    std::shared_ptr<parachain::ParachainStorage> pp_;
   };
 
 }  // namespace kagome::network

core/parachain/CMakeLists.txt

@@ -66,6 +66,7 @@ add_library(validator_parachain
     backing/store_impl.cpp
     backing/cluster.cpp
     validator/statement_distribution/statement_distribution.cpp
+    validator/parachain_storage.cpp
     )
 
 target_link_libraries(validator_parachain

core/parachain/approval/approval_distribution.cpp

@@ -577,7 +577,7 @@ namespace kagome::parachain {
       std::shared_ptr<crypto::KeyStore> keystore,
       std::shared_ptr<crypto::Hasher> hasher,
       std::shared_ptr<network::PeerView> peer_view,
-      std::shared_ptr<ParachainProcessorImpl> parachain_processor,
+      std::shared_ptr<ParachainProcessor> parachain_processor,
       std::shared_ptr<crypto::Sr25519Provider> crypto_provider,
       std::shared_ptr<network::PeerManager> pm,
       std::shared_ptr<network::Router> router,

core/parachain/approval/approval_distribution.hpp

@@ -59,7 +59,7 @@ namespace kagome::consensus::babe {
 
 namespace kagome::parachain {
   class ApprovalThreadPool;
-  class ParachainProcessorImpl;
+  class ParachainProcessor;
   class Pvf;
 }  // namespace kagome::parachain
 
@@ -299,7 +299,7 @@ namespace kagome::parachain {
         std::shared_ptr<crypto::KeyStore> keystore,
         std::shared_ptr<crypto::Hasher> hasher,
         std::shared_ptr<network::PeerView> peer_view,
-        std::shared_ptr<ParachainProcessorImpl> parachain_processor,
+        std::shared_ptr<ParachainProcessor> parachain_processor,
         std::shared_ptr<crypto::Sr25519Provider> crypto_provider,
         std::shared_ptr<network::PeerManager> pm,
         std::shared_ptr<network::Router> router,
@@ -876,7 +876,7 @@ namespace kagome::parachain {
           StorePair<Hash, DistribBlockEntry>>
         store_;
 
-    std::shared_ptr<ParachainProcessorImpl> parachain_processor_;
+    std::shared_ptr<ParachainProcessor> parachain_processor_;
     std::shared_ptr<crypto::Sr25519Provider> crypto_provider_;
     std::shared_ptr<network::PeerManager> pm_;
     std::shared_ptr<network::Router> router_;

core/parachain/validator/i_parachain_processor.hpp

@@ -7,9 +7,32 @@
 #pragma once
 
 #include "outcome/outcome.hpp"
+#include "network/types/collator_messages_vstaging.hpp"
+#include "parachain/validator/statement_distribution/types.hpp"
 
 namespace kagome::parachain {
 
+  class ParachainStorage {
+  public:
+  virtual ~ParachainStorage() = default;
+
+    virtual outcome::result<network::FetchChunkResponse> OnFetchChunkRequest(
+        const network::FetchChunkRequest &request) = 0;
+
+    virtual outcome::result<network::FetchChunkResponseObsolete>
+    OnFetchChunkRequestObsolete(const network::FetchChunkRequest &request) = 0;
+
+    /**
+     * @brief Fetches the Proof of Validity (PoV) for a given candidate.
+     *
+     * @param candidate_hash The hash of the candidate for which the PoV is to
+     * be fetched.
+     * @return network::ResponsePov The PoV associated with the given candidate
+     * hash.
+     */
+    virtual network::ResponsePov getPov(CandidateHash &&candidate_hash) = 0;
+  };
+
   class ParachainProcessor {
    public:
     enum class Error {
@@ -48,6 +71,23 @@ namespace kagome::parachain {
 
     virtual ~ParachainProcessor() = default;
 
+    virtual void onValidationProtocolMsg(
+        const libp2p::peer::PeerId &peer_id,
+        const network::VersionedValidatorProtocolMessage &message) = 0;
+
+    virtual void handle_advertisement(
+        const RelayHash &relay_parent,
+        const libp2p::peer::PeerId &peer_id,
+        std::optional<std::pair<CandidateHash, Hash>> &&prospective_candidate) = 0;
+
+    virtual void onIncomingCollator(const libp2p::peer::PeerId &peer_id,
+                            network::CollatorPublicKey pubkey,
+                            network::ParachainId para_id) = 0;
+
+    virtual outcome::result<void> canProcessParachains() const = 0;
+
+    virtual void handleStatement(const primitives::BlockHash &relay_parent,
+                                 const SignedFullStatementWithPVD &statement) = 0;
   };
 
 }

core/parachain/validator/impl/parachain_observer_impl.cpp

@@ -20,12 +20,14 @@ namespace kagome::parachain {
   ParachainObserverImpl::ParachainObserverImpl(
       std::shared_ptr<network::PeerManager> pm,
       std::shared_ptr<crypto::Sr25519Provider> crypto_provider,
-      std::shared_ptr<parachain::ParachainProcessorImpl> processor,
+      std::shared_ptr<parachain::ParachainProcessor> processor,
+      std::shared_ptr<parachain::ParachainStorage> parachain_storage,
       std::shared_ptr<network::PeerView> peer_view,
       std::shared_ptr<parachain::ApprovalDistribution> approval_distribution)
       : pm_{std::move(pm)},
         crypto_provider_{std::move(crypto_provider)},
         processor_{std::move(processor)},
+        parachain_storage_(std::move(parachain_storage)),
         peer_view_(std::move(peer_view)),
         approval_distribution_(std::move(approval_distribution)),
         logger_(log::createLogger("ParachainObserver", "parachain")) {
@@ -34,6 +36,7 @@ namespace kagome::parachain {
     BOOST_ASSERT_MSG(processor_, "Parachain processor must be initialized!");
     BOOST_ASSERT(peer_view_);
     BOOST_ASSERT(approval_distribution_);
+    BOOST_ASSERT(parachain_storage_);
   }
 
   void ParachainObserverImpl::onIncomingMessage(
@@ -106,7 +109,7 @@ namespace kagome::parachain {
   outcome::result<network::ResponsePov> ParachainObserverImpl::OnPovRequest(
       network::RequestPov request) {
     // NOLINTNEXTLINE(hicpp-move-const-arg,performance-move-const-arg)
-    return processor_->getPov(std::move(request));
+    return parachain_storage_->getPov(std::move(request));
   }
 
   outcome::result<network::CollationFetchingResponse>

core/parachain/validator/impl/parachain_observer_impl.hpp

@@ -21,7 +21,8 @@ namespace kagome::crypto {
 }
 
 namespace kagome::parachain {
-  class ParachainProcessorImpl;
+  class ParachainProcessor;
+  class ParachainStorage;
   class ApprovalDistribution;
 }  // namespace kagome::parachain
 
@@ -32,7 +33,8 @@ namespace kagome::parachain {
     ParachainObserverImpl(
         std::shared_ptr<network::PeerManager> pm,
         std::shared_ptr<crypto::Sr25519Provider> crypto_provider,
-        std::shared_ptr<parachain::ParachainProcessorImpl> processor,
+        std::shared_ptr<parachain::ParachainProcessor> processor,
+        std::shared_ptr<parachain::ParachainStorage> parachain_storage,
         std::shared_ptr<network::PeerView> peer_view,
         std::shared_ptr<parachain::ApprovalDistribution> approval_distribution);
 
@@ -71,7 +73,8 @@ namespace kagome::parachain {
 
     std::shared_ptr<network::PeerManager> pm_;
     std::shared_ptr<crypto::Sr25519Provider> crypto_provider_;
-    std::shared_ptr<parachain::ParachainProcessorImpl> processor_;
+    std::shared_ptr<parachain::ParachainProcessor> processor_;
+    std::shared_ptr<parachain::ParachainStorage> parachain_storage_;
     std::shared_ptr<network::PeerView> peer_view_;
     std::shared_ptr<parachain::ApprovalDistribution> approval_distribution_;
 

core/parachain/validator/impl/parachain_processor.cpp

@@ -83,7 +83,7 @@ namespace kagome::parachain {
       LazySPtr<consensus::SlotsUtil> slots_util,
       std::shared_ptr<consensus::babe::BabeConfigRepository> babe_config_repo,
       std::shared_ptr<statement_distribution::IStatementDistribution> sd)
-      : pm_(std::move(pm)),
+      : ParachainStorageImpl(std::move(av_store)), pm_(std::move(pm)),
         crypto_provider_(std::move(crypto_provider)),
         router_(std::move(router)),
         hasher_(std::move(hasher)),
@@ -94,7 +94,6 @@ namespace kagome::parachain {
         pvf_precheck_(std::move(pvf_precheck)),
         bitfield_store_(std::move(bitfield_store)),
         backing_store_(std::move(backing_store)),
-        av_store_(std::move(av_store)),
         parachain_host_(std::move(parachain_host)),
         app_config_(app_config),
         sync_state_observable_(std::move(sync_state_observable)),
@@ -116,7 +115,6 @@ namespace kagome::parachain {
     BOOST_ASSERT(bitfield_store_);
     BOOST_ASSERT(backing_store_);
     BOOST_ASSERT(pvf_);
-    BOOST_ASSERT(av_store_);
     BOOST_ASSERT(parachain_host_);
     BOOST_ASSERT(signer_factory_);
     BOOST_ASSERT(sync_state_observable_);
@@ -1189,37 +1187,6 @@ namespace kagome::parachain {
     }
   }
 
-  outcome::result<network::FetchChunkResponse>
-  ParachainProcessorImpl::OnFetchChunkRequest(
-      const network::FetchChunkRequest &request) {
-    if (auto chunk =
-            av_store_->getChunk(request.candidate, request.chunk_index)) {
-      return network::Chunk{
-          .data = chunk->chunk,
-          .chunk_index = request.chunk_index,
-          .proof = chunk->proof,
-      };
-    }
-    return network::Empty{};
-  }
-
-  outcome::result<network::FetchChunkResponseObsolete>
-  ParachainProcessorImpl::OnFetchChunkRequestObsolete(
-      const network::FetchChunkRequest &request) {
-    if (auto chunk =
-            av_store_->getChunk(request.candidate, request.chunk_index)) {
-      // This check needed because v1 protocol mustn't have chunk mapping
-      // https://github.com/paritytech/polkadot-sdk/blob/d2fd53645654d3b8e12cbf735b67b93078d70113/polkadot/node/core/av-store/src/lib.rs#L1345
-      if (chunk->index == request.chunk_index) {
-        return network::ChunkObsolete{
-            .data = chunk->chunk,
-            .proof = chunk->proof,
-        };
-      }
-    }
-    return network::Empty{};
-  }
-
   std::optional<
       std::reference_wrapper<ParachainProcessorImpl::RelayParentState>>
   ParachainProcessorImpl::tryGetStateByRelayParent(
@@ -2063,14 +2030,6 @@ namespace kagome::parachain {
     return sign_result.value();
   }
 
-  network::ResponsePov ParachainProcessorImpl::getPov(
-      CandidateHash &&candidate_hash) {
-    if (auto res = av_store_->getPov(candidate_hash)) {
-      return network::ResponsePov{*res};
-    }
-    return network::Empty{};
-  }
-
   void ParachainProcessorImpl::onIncomingCollator(
       const libp2p::peer::PeerId &peer_id,
       network::CollatorPublicKey pubkey,