fix deals (#471)

turuslan · web-flow · commit 399ec8695add · 2021-09-28T21:06:10.000+03:00
Signed-off-by: turuslan &lt;turuslan.devbox@gmail.com&gt;
diff --git a/core/api/full_node/make.cpp b/core/api/full_node/make.cpp
@@ -22,6 +22,7 @@
 #include "primitives/tipset/chain.hpp"
 #include "proofs/impl/proof_engine_impl.hpp"
 #include "storage/car/car.hpp"
+#include "storage/unixfs/unixfs.hpp"
 #include "vm/actor/builtin/states/account/account_actor_state.hpp"
 #include "vm/actor/builtin/states/init/init_actor_state.hpp"
 #include "vm/actor/builtin/states/market/market_actor_state.hpp"
@@ -458,9 +459,6 @@ namespace fc::api {
      * Initiates the retrieval deal of a file in retrieval market.
      */
     api->ClientRetrieve = [=](auto &&cb, auto order, auto &&file_ref) {
-      if (!file_ref.is_car) {
-        return cb(ERROR_TEXT("ClientRetrieve unixfs not implemented"));
-      }
       if (order.size == 0) {
         return cb(ERROR_TEXT("Cannot make retrieval deal for zero bytes"));
       }
@@ -491,8 +489,14 @@ namespace fc::api {
               return cb(res.error());
             }
             logger->info("retrieval deal done");
-            OUTCOME_CB1(storage::car::makeSelectiveCar(
-                *markets_ipld, {{order.root, {}}}, file_ref.path));
+            if (file_ref.is_car) {
+              OUTCOME_CB1(storage::car::makeSelectiveCar(
+                  *markets_ipld, {{order.root, {}}}, file_ref.path));
+            } else {
+              std::ofstream file{file_ref.path};
+              OUTCOME_CB1(
+                  storage::unixfs::unwrapFile(file, *markets_ipld, order.root));
+            }
             cb(outcome::success());
           }));
     };
diff --git a/core/data_transfer/dt.cpp b/core/data_transfer/dt.cpp
@@ -81,7 +81,7 @@ namespace fc::data_transfer {
                 push.on_begin = {};
                 if (res.is_accepted) {
                   storage::ipld::traverser::Traverser t{
-                      *push.ipld, req.root_cid, CborRaw{req.selector}};
+                      *push.ipld, req.root_cid, CborRaw{req.selector}, true};
                   gsns::Response res{
                       gsns::ResponseStatusCode::RS_FULL_CONTENT, {}, {}};
                   if (auto _cids{t.traverseAll()}) {
diff --git a/core/markets/retrieval/client/impl/retrieval_client_impl.cpp b/core/markets/retrieval/client/impl/retrieval_client_impl.cpp
@@ -9,6 +9,20 @@
 #include "common/libp2p/peer/peer_info_helper.hpp"
 
 namespace fc::markets::retrieval::client {
+  DealState::DealState(const DealProposal &proposal,
+                       const IpldPtr &ipld,
+                       const RetrieveResponseHandler &handler,
+                       const Address &client_wallet,
+                       const Address &miner_wallet,
+                       const TokenAmount &total_funds)
+      : proposal{proposal},
+        state{proposal.params},
+        handler{handler},
+        client_wallet{client_wallet},
+        miner_wallet{miner_wallet},
+        total_funds(total_funds),
+        traverser{
+            *ipld, proposal.payload_cid, proposal.params.selector, false} {}
 
   RetrievalClientImpl::RetrievalClientImpl(
       std::shared_ptr<Host> host,
diff --git a/core/markets/retrieval/client/impl/retrieval_client_impl.hpp b/core/markets/retrieval/client/impl/retrieval_client_impl.hpp
@@ -38,14 +38,7 @@ namespace fc::markets::retrieval::client {
               const RetrieveResponseHandler &handler,
               const Address &client_wallet,
               const Address &miner_wallet,
-              const TokenAmount &total_funds)
-        : proposal{proposal},
-          state{proposal.params},
-          handler{handler},
-          client_wallet{client_wallet},
-          miner_wallet{miner_wallet},
-          total_funds(total_funds),
-          traverser{*ipld, proposal.payload_cid, proposal.params.selector} {}
+              const TokenAmount &total_funds);
 
     DealProposal proposal;
     State state;
diff --git a/core/markets/retrieval/provider/impl/retrieval_provider_impl.cpp b/core/markets/retrieval/provider/impl/retrieval_provider_impl.cpp
@@ -191,6 +191,7 @@ namespace fc::markets::retrieval::provider {
             *deal->ipld,
             deal->proposal.payload_cid,
             deal->proposal.params.selector,
+            false,
         });
         deal->unsealed = true;
         doBlocks(deal);
diff --git a/core/markets/storage/client/impl/storage_market_client_impl.cpp b/core/markets/storage/client/impl/storage_market_client_impl.cpp
@@ -336,39 +336,6 @@ namespace fc::markets::storage::client {
     return client_deal->proposal_cid;
   }
 
-  outcome::result<StorageParticipantBalance>
-  StorageMarketClientImpl::getPaymentEscrow(const Address &address) const {
-    OUTCOME_TRY(chain_head, api_->ChainHead());
-    OUTCOME_TRY(balance, api_->StateMarketBalance(address, chain_head->key));
-    return StorageParticipantBalance{balance.locked,
-                                     balance.escrow - balance.locked};
-  }
-
-  outcome::result<void> StorageMarketClientImpl::addPaymentEscrow(
-      const Address &address, const TokenAmount &amount) {
-    UnsignedMessage unsigned_message{
-        kStorageMarketAddress,
-        address,
-        {},
-        amount,
-        kDefaultGasPrice,
-        kDefaultGasLimit,
-        vm::actor::builtin::v0::market::AddBalance::Number,
-        {}};
-    OUTCOME_TRY(signed_message,
-                api_->MpoolPushMessage(unsigned_message, api::kPushNoSpec));
-    // TODO: maybe async call, it's long
-    OUTCOME_TRY(msg_state,
-                api_->StateWaitMsg(signed_message.getCid(),
-                                   kMessageConfidence,
-                                   api::kLookbackNoLimit,
-                                   true));
-    if (msg_state.receipt.exit_code != VMExitCode::kOk) {
-      return StorageMarketClientError::kAddFundsCallError;
-    }
-    return outcome::success();
-  }
-
   outcome::result<SignedStorageAsk>
   StorageMarketClientImpl::validateAskResponse(
       const outcome::result<AskResponse> &response,
diff --git a/core/markets/storage/client/impl/storage_market_client_impl.hpp b/core/markets/storage/client/impl/storage_market_client_impl.hpp
@@ -84,12 +84,6 @@ namespace fc::markets::storage::client {
         bool verified_deal,
         bool is_fast_retrieval) override;
 
-    outcome::result<StorageParticipantBalance> getPaymentEscrow(
-        const Address &address) const override;
-
-    outcome::result<void> addPaymentEscrow(const Address &address,
-                                           const TokenAmount &amount) override;
-
    private:
     void askDealStatus(std::shared_ptr<ClientDeal> deal);
 
diff --git a/core/markets/storage/client/import_manager/import_manager.cpp b/core/markets/storage/client/import_manager/import_manager.cpp
@@ -30,31 +30,30 @@ namespace fc::markets::storage::client::import_manager {
   outcome::result<CID> ImportManager::import(
       const boost::filesystem::path &path, bool is_car) {
     CID root;
+    std::shared_ptr<MemoryIndexedCar> ipld;
+    std::string tmp_path;
     if (is_car) {
-      // validate CAR
-      OUTCOME_TRY(ipld, MemoryIndexedCar::make(path.string(), false));
+      OUTCOME_TRYA(ipld, MemoryIndexedCar::make(path.string(), false));
       if (ipld->roots.size() != 1) {
         return ERROR_TEXT(
             "StorageMarketImportManager: cannot import car with more that one "
             "root");
       }
       root = ipld->roots.front();
-      OUTCOME_TRY(car_path, makeFilename(root));
-      boost::filesystem::copy_file(
-          path, car_path, boost::filesystem::copy_option::overwrite_if_exists);
-      padPiece(car_path);
     } else {
-      const auto tmp_path{path.string() + ".unixfs-tmp.car"};
-      OUTCOME_TRY(ipld, MemoryIndexedCar::make(tmp_path, true));
+      tmp_path = path.string() + ".unixfs-tmp.car";
+      OUTCOME_TRYA(ipld, MemoryIndexedCar::make(tmp_path, true));
       std::ifstream file{path.string()};
       OUTCOME_TRYA(root, wrapFile(*ipld, file));
-      OUTCOME_TRY(car_path, makeFilename(root));
-      // add header with root and reorder objects
-      OUTCOME_TRY(::fc::storage::car::makeSelectiveCar(
-          *ipld, {{root, {}}}, car_path.string()));
+    }
+    OUTCOME_TRY(car_path, makeFilename(root));
+    // add header with root and reorder objects
+    OUTCOME_TRY(::fc::storage::car::makeSelectiveCar(
+        *ipld, {{root, {}}}, car_path.string()));
+    if (!tmp_path.empty()) {
       boost::filesystem::remove(tmp_path);
-      padPiece(car_path);
     }
+    padPiece(car_path);
     OUTCOME_TRY(addImported(root, path));
     return root;
   }
diff --git a/core/markets/storage/client/storage_market_client.hpp b/core/markets/storage/client/storage_market_client.hpp
@@ -96,12 +96,6 @@ namespace fc::markets::storage::client {
         const RegisteredSealProof &registered_proof,
         bool verified_deal,
         bool is_fast_retrieval) = 0;
-
-    virtual outcome::result<StorageParticipantBalance> getPaymentEscrow(
-        const Address &address) const = 0;
-
-    virtual outcome::result<void> addPaymentEscrow(
-        const Address &address, const TokenAmount &amount) = 0;
   };
 
 }  // namespace fc::markets::storage::client
diff --git a/core/node/graphsync_server.cpp b/core/node/graphsync_server.cpp
@@ -27,7 +27,7 @@ namespace fc::sync {
                    common::hex_lower(request.selector));
 
       storage::ipld::traverser::Traverser traverser{
-          ipld, request.root_cid, {request.selector}};
+          ipld, request.root_cid, {request.selector}, true};
       auto ok{true};
       if (auto _cids{traverser.traverseAll()}) {
         for (auto &cid : _cids.value()) {
diff --git a/core/storage/car/car.cpp b/core/storage/car/car.cpp
@@ -126,7 +126,7 @@ namespace fc::storage::car {
   outcome::result<Buffer> makeCar(Ipld &store, const std::vector<CID> &roots) {
     std::set<CID> cids;
     for (auto &root : roots) {
-      Traverser traverser{store, root, kAllSelector};
+      Traverser traverser{store, root, kAllSelector, true};
       OUTCOME_TRY(visited, traverser.traverseAll());
       cids.insert(visited.begin(), visited.end());
     }
@@ -139,7 +139,7 @@ namespace fc::storage::car {
     std::vector<CID> cid_order;
     std::set<CID> cids;
     for (auto &dag : dags) {
-      Traverser traverser{store, dag.first, dag.second};
+      Traverser traverser{store, dag.first, dag.second, true};
       OUTCOME_TRY(visited, traverser.traverseAll());
       roots.push_back(dag.first);
       for (auto &cid : visited) {
@@ -201,7 +201,7 @@ namespace fc::storage::car {
     std::vector<CID> cid_order;
     std::set<CID> cids;
     for (auto &dag : dags) {
-      Traverser traverser{store, dag.first, dag.second};
+      Traverser traverser{store, dag.first, dag.second, true};
       OUTCOME_TRY(visited, traverser.traverseAll());
       roots.push_back(dag.first);
       for (auto &cid : visited) {
diff --git a/core/storage/ipfs/graphsync/impl/network/network_fwd.hpp b/core/storage/ipfs/graphsync/impl/network/network_fwd.hpp
@@ -97,7 +97,7 @@ namespace fc::storage::ipfs::graphsync {
   constexpr std::string_view kProtocolVersion = "/ipfs/graphsync/1.0.0";
 
   /// Max byte size of individual message
-  constexpr size_t kMaxMessageSize = 16 * 1024 * 1024;
+  constexpr size_t kMaxMessageSize = 4 * 1024 * 1024;
 
   /// Max byte size of pending message queue
   constexpr size_t kMaxPendingBytes = 64 * 1024 * 1024;
diff --git a/core/storage/ipfs/graphsync/impl/network/outbound_endpoint.cpp b/core/storage/ipfs/graphsync/impl/network/outbound_endpoint.cpp
@@ -48,7 +48,9 @@ namespace fc::storage::ipfs::graphsync {
         queue_ ? queue_->getState().pending_bytes : pending_bytes_;
 
     if (pending_bytes + data->size() > max_pending_bytes_) {
-      return Error::kWriteQueueOverflow;
+      logger()->warn("OutboundEndpoint::enqueue overflow {}/{}",
+                     pending_bytes + data->size(),
+                     max_pending_bytes_);
     }
 
     if (queue_) {
@@ -74,8 +76,7 @@ namespace fc::storage::ipfs::graphsync {
       return res.error();
     }
 
-    logger()->debug(
-        "enqueueing response, size=", res.value()->size());
+    logger()->debug("enqueueing response, size={}", res.value()->size());
 
     return enqueue(std::move(res.value()));
   }
@@ -87,7 +88,7 @@ namespace fc::storage::ipfs::graphsync {
     auto serialized_size = response_builder_.getSerializedSize();
 
     if (serialized_size + data.size() > kMaxMessageSize) {
-      static const Response partial{RS_PARTIAL_CONTENT, {}, {}};
+      static const Response partial{RS_PARTIAL_RESPONSE, {}, {}};
       OUTCOME_TRY(sendResponse(request_id, partial));
     }
 
diff --git a/core/storage/ipfs/graphsync/impl/network/peer_context.cpp b/core/storage/ipfs/graphsync/impl/network/peer_context.cpp
@@ -267,8 +267,8 @@ namespace fc::storage::ipfs::graphsync {
       outbound_endpoint_.reset();
     }
 
-    stream->close(
-        [stream](outcome::result<void>) { logger()->trace("stream closed"); });
+    stream->reset();
+    logger()->trace("stream closed");
   }
 
   void PeerContext::closeLocalRequests(ResponseStatusCode status) {
diff --git a/core/storage/ipld/traverser.cpp b/core/storage/ipld/traverser.cpp
@@ -54,6 +54,14 @@ namespace fc::storage::ipld::traverser {
     }
   };
 
+  Traverser::Traverser(Ipld &store,
+                       const CID &root,
+                       const Selector &selector,
+                       bool unique)
+      : store{store}, unique{unique} {
+    to_visit_.push(root);
+  }
+
   outcome::result<std::vector<CID>> Traverser::traverseAll() {
     // TODO(turuslan): implement selectors
     while (!isCompleted()) {
@@ -66,22 +74,29 @@ namespace fc::storage::ipld::traverser {
     if (isCompleted()) {
       return TraverserError::kTraverseCompleted;
     }
-    CID cid = to_visit_.front();
+    CID cid{std::move(to_visit_.front())};
     to_visit_.pop();
-    if (visited_.insert(cid).second) {
-      OUTCOME_TRY(bytes, store.get(cid));
-      visit_order_.push_back(cid);
-
-      // TODO(turuslan): what about other types?
-      if (cid.content_type == CID::Multicodec::DAG_CBOR) {
-        CborDecodeStream s{bytes};
-        OUTCOME_TRY(parseCbor(s));
-      } else if (cid.content_type == CID::Multicodec::DAG_PB) {
-        OUTCOME_TRY(cids, PbNodeDecoder::links(bytes));
-        for (auto &&c : cids) {
-          to_visit_.push(c);
+    if (unique) {
+      visited_.insert(cid);
+    }
+    auto BOOST_OUTCOME_TRY_UNIQUE_NAME{gsl::finally([&] {
+      if (unique) {
+        while (!to_visit_.empty() && visited_.count(to_visit_.front()) != 0) {
+          to_visit_.pop();
         }
       }
+    })};
+    OUTCOME_TRY(bytes, store.get(cid));
+    visit_order_.push_back(cid);
+    // TODO(turuslan): what about other types?
+    if (cid.content_type == CID::Multicodec::DAG_CBOR) {
+      CborDecodeStream s{bytes};
+      OUTCOME_TRY(parseCbor(s));
+    } else if (cid.content_type == CID::Multicodec::DAG_PB) {
+      OUTCOME_TRY(cids, PbNodeDecoder::links(bytes));
+      for (auto &&c : cids) {
+        to_visit_.push(c);
+      }
     }
     return cid;
   }
diff --git a/core/storage/ipld/traverser.hpp b/core/storage/ipld/traverser.hpp
@@ -29,11 +29,12 @@ namespace fc::storage::ipld::traverser {
      * @param store - ipld store
      * @param root - root cid
      * @param selector - selector
+     * @param unique - should skip duplicates
      */
-    Traverser(Ipld &store, const CID &root, const Selector &selector)
-        : store{store} {
-      to_visit_.push(root);
-    }
+    Traverser(Ipld &store,
+              const CID &root,
+              const Selector &selector,
+              bool unique);
 
     /**
      * Traverse all from the root
@@ -58,6 +59,7 @@ namespace fc::storage::ipld::traverser {
     outcome::result<void> parseCbor(CborDecodeStream &s);
 
     Ipld &store;
+    bool unique{};
     std::queue<CID> to_visit_;      // set of cids to visit
     std::vector<CID> visit_order_;  // visited cids in visit order
     std::set<CID> visited_;         // set of visited cids
diff --git a/core/storage/unixfs/CMakeLists.txt b/core/storage/unixfs/CMakeLists.txt
@@ -7,4 +7,5 @@ add_library(unixfs
 target_link_libraries(unixfs
     cbor
     filecoin_hasher
+    ipld_traverser
     )
diff --git a/core/storage/unixfs/unixfs.cpp b/core/storage/unixfs/unixfs.cpp
diff --git a/core/storage/unixfs/unixfs.hpp b/core/storage/unixfs/unixfs.hpp
diff --git a/test/core/storage/unixfs/unixfs_test.cpp b/test/core/storage/unixfs/unixfs_test.cpp
