Add DataTransfer protocol (#144)

Signed-off-by: Alexey-N-Chernyshov <[email protected]>

Author: Alexey-N-Chernyshov

cmake/Hunter/config.cmake

@@ -36,7 +36,7 @@ hunter_config(
 )
 
 hunter_config(libp2p
-    URL https://github.com/soramitsu/libp2p/archive/bf2aae97a9a00fe3493ee05d2c679e2aed790e28.zip
-    SHA1 e051b41120a67f0fb28f3af2cc357e34ad33de2b
-    CMAKE_ARGS TESTING=OFF
+    URL https://github.com/soramitsu/libp2p/archive/d65b00317c0c190f6f1870fce48284e6709994f0.zip
+    SHA1 4b2645f33a0f406c51d42b37a6521e946ffae2ef
+    CMAKE_ARGS TESTING=OFF EXPOSE_MOCKS=ON
     )

core/CMakeLists.txt

@@ -10,6 +10,7 @@ add_subdirectory(clock)
 add_subdirectory(codec)
 add_subdirectory(common)
 add_subdirectory(crypto)
+add_subdirectory(data_transfer)
 add_subdirectory(primitives)
 add_subdirectory(proofs)
 add_subdirectory(sectorbuilder)

core/adt/uvarint_key.cpp

@@ -26,9 +26,10 @@ namespace fc::adt {
 
   outcome::result<UvarintKeyer::Key> UvarintKeyer::decode(
       const std::string &key) {
-    // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast)
     auto maybe = UVarint::create(gsl::make_span(
-        reinterpret_cast<const uint8_t *>(key.data()), key.size()));
+        // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast)
+        reinterpret_cast<const uint8_t *>(key.data()),
+        key.size()));
     if (!maybe) {
       return UvarintKeyError::DECODE_ERROR;
     }

core/crypto/bls/impl/bls_provider_impl.cpp

@@ -91,6 +91,7 @@ namespace fc::crypto::bls {
   outcome::result<Signature> BlsProviderImpl::aggregateSignatures(
       const std::vector<Signature> &signatures) const {
     Signature signature;
+    // NOLINTNEXTLINE(cppcoreguidelines-pro-type-reinterpret-cast)
     const uint8_t *flat_bytes =
         reinterpret_cast<const uint8_t *>(signatures.data());
     size_t flat_size = sizeof(Signature) * signatures.size();

core/data_transfer/CMakeLists.txt

@@ -0,0 +1,16 @@
+#
+# Copyright Soramitsu Co., Ltd. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+
+add_subdirectory(impl)
+
+add_library(data_transfer
+    message_receiver.cpp
+    message.cpp
+    )
+target_link_libraries(data_transfer
+    p2p::p2p
+    logger
+    ipld_node
+    )

core/data_transfer/README.md

@@ -0,0 +1,4 @@
+Module to perform data transfer over graphsync.
+
+This module encapsulates protocols for exchanging `Piece` data between storage clients and miners, both when consummating 
+a storage deal and when retrieving the piece later.

core/data_transfer/impl/CMakeLists.txt

@@ -0,0 +1,15 @@
+#
+# Copyright Soramitsu Co., Ltd. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+
+add_subdirectory(graphsync)
+
+add_library(libp2p_data_transfer_network
+    libp2p_data_transfer_network.cpp
+    stream_message_sender.cpp
+    )
+target_link_libraries(libp2p_data_transfer_network
+    data_transfer
+    p2p::p2p
+    )

core/data_transfer/impl/graphsync/CMakeLists.txt

@@ -0,0 +1,15 @@
+#
+# Copyright Soramitsu Co., Ltd. All Rights Reserved.
+# SPDX-License-Identifier: Apache-2.0
+#
+
+add_library(data_transfer_graphsync
+    graphsync_manager.cpp
+    graphsync_receiver.cpp
+    data_transfer_extension.cpp
+    )
+target_link_libraries(data_transfer_graphsync
+    clock
+    graphsync
+    libp2p_data_transfer_network
+    )

core/data_transfer/impl/graphsync/data_transfer_extension.cpp

@@ -0,0 +1,44 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "data_transfer_extension.hpp"
+#include "codec/cbor/cbor.hpp"
+
+namespace fc::data_transfer {
+
+  using libp2p::peer::PeerId;
+
+  PeerId ExtensionDataTransferData::getInitiator() const {
+    return PeerId::fromBase58(initiator).value();
+  }
+
+  outcome::result<Extension> encodeDataTransferExtension(
+      const ExtensionDataTransferData &data) {
+    OUTCOME_TRY(bytes, codec::cbor::encode(data));
+    return Extension{.name = std::string(kDataTransferExtensionName),
+                     .data = bytes.toVector()};
+  }
+
+  /// Decodes Data Transfer graphsync extension
+  outcome::result<ExtensionDataTransferData> decodeDataTransferExtension(
+      const Extension &extension) {
+    if (extension.name != kDataTransferExtensionName) {
+      return DataTransferExtensionError::UNEXPECTED_EXTENSION_NAME;
+    }
+    return codec::cbor::decode<ExtensionDataTransferData>(extension.data);
+  }
+
+}  // namespace fc::data_transfer
+
+OUTCOME_CPP_DEFINE_CATEGORY(fc::data_transfer, DataTransferExtensionError, e) {
+  using fc::data_transfer::DataTransferExtensionError;
+
+  switch (e) {
+    case DataTransferExtensionError::UNEXPECTED_EXTENSION_NAME:
+      return "DataTransferExtensionError: unexpected extension name";
+  }
+
+  return "unknown error";
+}

core/data_transfer/impl/graphsync/data_transfer_extension.hpp

@@ -0,0 +1,50 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#ifndef CPP_FILECOIN_DATA_TRANSFER_EXTENSION_HPP
+#define CPP_FILECOIN_DATA_TRANSFER_EXTENSION_HPP
+
+#include <libp2p/peer/peer_id.hpp>
+#include "data_transfer/types.hpp"
+#include "storage/ipfs/graphsync/extension.hpp"
+
+namespace fc::data_transfer {
+
+  using libp2p::peer::PeerId;
+  using storage::ipfs::graphsync::Extension;
+
+  constexpr std::string_view kDataTransferExtensionName = "fil/data-transfer";
+
+  /**
+   * Data Transfer extension for graphsync protocol
+   */
+  struct ExtensionDataTransferData {
+    TransferId transfer_id;
+    std::string initiator;
+    bool is_pull;
+
+    PeerId getInitiator() const;
+  };
+
+  CBOR_TUPLE(ExtensionDataTransferData, transfer_id, initiator, is_pull)
+
+  /// Encodes Data Transfer graphsync extension
+  outcome::result<Extension> encodeDataTransferExtension(
+      const ExtensionDataTransferData &data);
+
+  /// Decodes Data Transfer graphsync extension
+  outcome::result<ExtensionDataTransferData> decodeDataTransferExtension(
+      const Extension &extension);
+
+  /**
+   * @brief Type of errors returned by ExtensionDataTransferData
+   */
+  enum class DataTransferExtensionError { UNEXPECTED_EXTENSION_NAME };
+
+}  // namespace fc::data_transfer
+
+OUTCOME_HPP_DECLARE_ERROR(fc::data_transfer, DataTransferExtensionError);
+
+#endif  // CPP_FILECOIN_DATA_TRANSFER_EXTENSION_HPP

core/data_transfer/impl/graphsync/graphsync_manager.cpp

@@ -0,0 +1,133 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#include "data_transfer/impl/graphsync/graphsync_manager.hpp"
+
+#include <utility>
+#include "data_transfer/message.hpp"
+
+namespace fc::data_transfer::graphsync {
+
+  using common::Buffer;
+
+  GraphSyncManager::GraphSyncManager(std::shared_ptr<Host> host,
+                                     PeerId peer)
+      : network_(std::move(host)), peer_(std::move(peer)) {}
+
+  outcome::result<ChannelId> GraphSyncManager::openPushDataChannel(
+      const PeerId &to,
+      const Voucher &voucher,
+      CID base_cid,
+      std::shared_ptr<IPLDNode> selector) {
+    OUTCOME_TRY(transfer_id,
+                sendDtRequest(selector, false, voucher, base_cid, to));
+    // initiator = us, sender = us, receiver = them
+    OUTCOME_TRY(
+        channel_id,
+        createChannel(
+            transfer_id, base_cid, selector, voucher.bytes, peer_, peer_, to));
+    return std::move(channel_id);
+  }
+
+  outcome::result<ChannelId> GraphSyncManager::openPullDataChannel(
+      const PeerId &to,
+      const Voucher &voucher,
+      CID base_cid,
+      std::shared_ptr<IPLDNode> selector) {
+    OUTCOME_TRY(transfer_id,
+                sendDtRequest(selector, true, voucher, base_cid, to));
+    // initiator = us, sender = them, receiver = us
+    OUTCOME_TRY(
+        channel_id,
+        createChannel(
+            transfer_id, base_cid, selector, voucher.bytes, peer_, to, peer_));
+    return std::move(channel_id);
+  }
+
+  outcome::result<ChannelId> GraphSyncManager::createChannel(
+      const TransferId &transfer_id,
+      const CID &base_cid,
+      std::shared_ptr<IPLDNode> selector,
+      const std::vector<uint8_t> &voucher,
+      const PeerId &initiator,
+      const PeerId &sender_peer,
+      const PeerId &receiver_peer) {
+    ChannelId channel_id{.initiator = initiator, .id = transfer_id};
+    Channel channel{.transfer_id = 0,
+                    .base_cid = base_cid,
+                    .selector = std::move(selector),
+                    .voucher = voucher,
+                    .sender = sender_peer,
+                    .recipient = receiver_peer,
+                    .total_size = 0};
+    ChannelState state{.channel = channel, .sent = 0, .received = 0};
+
+    // TODO check thread-safety of channels_
+    auto res = channels_.try_emplace(channel_id, state);
+    if (!res.second) return GraphsyncManagerError::STATE_ALREADY_EXISTS;
+
+    return channel_id;
+  }
+
+  outcome::result<void> GraphSyncManager::closeChannel(
+      const ChannelId &channel_id) {
+    return outcome::success();
+  }
+
+  outcome::result<TransferId> GraphSyncManager::sendDtRequest(
+      const std::shared_ptr<IPLDNode> &selector,
+      bool is_pull,
+      const Voucher &voucher,
+      const CID &base_cid,
+      const PeerId &to) {
+    std::vector<uint8_t> selector_bytes = selector->getRawBytes().toVector();
+
+    TransferId tx_id = ++last_tx_id;
+    OUTCOME_TRY(base_cid_str, base_cid.toString());
+    DataTransferMessage message = createRequest(base_cid_str,
+                                                is_pull,
+                                                selector_bytes,
+                                                voucher.bytes,
+                                                voucher.type,
+                                                tx_id);
+    OUTCOME_TRY(sender, network_.newMessageSender(to));
+    OUTCOME_TRY(sender->sendMessage(message));
+    return tx_id;
+  }
+
+  boost::optional<ChannelState> GraphSyncManager::getChannelByIdAndSender(
+      const ChannelId &channel_id, const PeerId &sender) {
+    // TODO check thread-safety of channels_
+    auto found = channels_.find(channel_id);
+    if (found == channels_.end() || found->second.channel.sender != sender) {
+      return boost::none;
+    }
+
+    return found->second;
+  }
+
+  outcome::result<void> GraphSyncManager::sendResponse(bool is_accepted,
+                                                       const PeerId &to,
+                                                       TransferId transfer_id) {
+    DataTransferMessage message = createResponse(is_accepted, transfer_id);
+    OUTCOME_TRY(sender, network_.newMessageSender(to));
+    OUTCOME_TRY(sender->sendMessage(message));
+    return outcome::success();
+  }
+
+}  // namespace fc::data_transfer::graphsync
+
+OUTCOME_CPP_DEFINE_CATEGORY(fc::data_transfer::graphsync,
+                            GraphsyncManagerError,
+                            e) {
+  using fc::data_transfer::graphsync::GraphsyncManagerError;
+
+  switch (e) {
+    case GraphsyncManagerError::STATE_ALREADY_EXISTS:
+      return "GraphsyncManagerError: state already exists";
+  }
+
+  return "unknown error";
+}