Miner fix (#475)

* Miner fix

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

* Update test

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

Author: ortyomka

core/api/full_node/make.cpp

@@ -351,8 +351,11 @@ namespace fc::api {
       return ts_load->load(tipset_key);
     };
     api->ChainGetTipSetByHeight =
-        [=](auto height, auto &tipset_key) -> outcome::result<TipsetCPtr> {
+        [=](auto height, auto tipset_key) -> outcome::result<TipsetCPtr> {
       std::shared_lock ts_lock{*env_context.ts_branches_mutex};
+      if (tipset_key.cids().empty()) {
+        tipset_key = chain_store->heaviestTipset()->key;
+      }
       OUTCOME_TRY(ts_branch, TsBranch::make(ts_load, tipset_key, ts_main));
       OUTCOME_TRY(it, find(ts_branch, height));
       return ts_load->lazyLoad(it.second->second);

core/api/rpc/json.hpp

@@ -15,6 +15,7 @@
 #include "api/worker_api.hpp"
 #include "common/enum.hpp"
 #include "common/libp2p/peer/cbor_peer_info.hpp"
+#include "miner/main/type.hpp"
 #include "payment_channel_manager/payment_channel_manager.hpp"
 #include "primitives/address/address_codec.hpp"
 #include "primitives/cid/cid_of_cbor.hpp"
@@ -40,12 +41,15 @@ namespace fc::codec::cbor {
 }  // namespace fc::codec::cbor
 
 namespace fc::api {
+  using api::ApiSectorInfo;
   using codec::cbor::CborDecodeStream;
   using common::Blob;
   using crypto::signature::BlsSignature;
   using crypto::signature::Secp256k1Signature;
   using crypto::signature::Signature;
   using markets::storage::StorageAsk;
+  using miner::types::PreSealSector;
+  using mining::SealingState;
   using primitives::BigInt;
   using primitives::FsStat;
   using primitives::kChainEpochUndefined;
@@ -245,6 +249,14 @@ namespace fc::api {
       decodeEnum(v, j);
     }
 
+    ENCODE(SealingState) {
+      return encode(common::to_int(v));
+    }
+
+    DECODE(SealingState) {
+      decodeEnum(v, j);
+    }
+
     ENCODE(PathType) {
       return encode(common::to_int(v));
     }
@@ -1012,6 +1024,16 @@ namespace fc::api {
       Get(j, "SealedCID", v.sealed_cid);
     }
 
+    ENCODE(ApiSectorInfo) {
+      Value j{rapidjson::kObjectType};
+      Set(j, "State", v.state);
+      return j;
+    }
+
+    DECODE(ApiSectorInfo) {
+      Get(j, "State", v.state);
+    }
+
     ENCODE(PowerPair) {
       Value j{rapidjson::kObjectType};
       Set(j, "Raw", v.raw);
@@ -1414,6 +1436,48 @@ namespace fc::api {
       decode(v.verified_deal, Get(j, "VerifiedDeal"));
     }
 
+    ENCODE(PreSealSector) {
+      Value j{rapidjson::kObjectType};
+      Set(j, "CommR", v.comm_r);
+      Set(j, "CommD", v.comm_d);
+      Set(j, "SectorID", v.sector_id);
+      Set(j, "Deal", v.deal);
+      Set(j, "ProofType", v.proof_type);
+      return j;
+    }
+
+    DECODE(PreSealSector) {
+      decode(v.comm_r, Get(j, "CommR"));
+      decode(v.comm_d, Get(j, "CommD"));
+      decode(v.sector_id, Get(j, "SectorID"));
+      decode(v.deal, Get(j, "Deal"));
+      decode(v.proof_type, Get(j, "ProofType"));
+    }
+
+    ENCODE(miner::types::Miner) {
+      Value j{rapidjson::kObjectType};
+      Set(j, "ID", v.id);
+      Set(j, "Owner", v.owner);
+      Set(j, "Worker", v.worker);
+      Set(j, "PeerId", v.peer_id);
+      Set(j, "MarketBalance", v.market_balance);
+      Set(j, "PowerBalance", v.power_balance);
+      Set(j, "SectorSize", v.sector_size);
+      Set(j, "Sectors", v.sectors);
+      return j;
+    }
+
+    DECODE(miner::types::Miner) {
+      decode(v.id, Get(j, "ID"));
+      decode(v.owner, Get(j, "Owner"));
+      decode(v.worker, Get(j, "Worker"));
+      decode(v.peer_id, Get(j, "PeerId"));
+      decode(v.market_balance, Get(j, "MarketBalance"));
+      decode(v.power_balance, Get(j, "PowerBalance"));
+      decode(v.sector_size, Get(j, "SectorSize"));
+      decode(v.sectors, Get(j, "Sectors"));
+    }
+
     template <typename T>
     ENCODE(Chan<T>) {
       return encode(v.id);

core/api/rpc/wsc.cpp

@@ -136,7 +136,8 @@ namespace fc::api::rpc {
           }
           if (id) {
             boost::asio::post(
-                io2, [this, close, id{*id}, value{std::move(value)}]() mutable {
+                *thread_chan.io,
+                [this, close, id{*id}, value{std::move(value)}]() mutable {
                   std::unique_lock lock{mutex};
                   auto it{chans.find(id)};
                   if (it != chans.end()) {

core/api/rpc/wsc.hpp

@@ -18,6 +18,7 @@
 #include "api/rpc/json.hpp"
 #include "api/rpc/rpc.hpp"
 #include "api/visit.hpp"
+#include "common/io_thread.hpp"
 #include "common/logger.hpp"
 
 namespace fc::api::rpc {
@@ -53,6 +54,7 @@ namespace fc::api::rpc {
 
    private:
     std::thread thread;
+    IoThread thread_chan;
     io_context io;
     io_context &io2;
     boost::asio::executor_work_guard<io_context::executor_type> work_guard;

core/api/storage_miner/storage_api.cpp

@@ -68,6 +68,24 @@ namespace fc::api {
       return outcome::success();
     };
 
+    api->SectorsList = [=]() -> outcome::result<std::vector<SectorNumber>> {
+      std::vector<SectorNumber> result;
+      for (const auto &sector : miner->getSealing()->getListSectors()) {
+        if (sector->state != mining::SealingState::kStateUnknown) {
+          result.push_back(sector->sector_number);
+        }
+      }
+
+      return result;
+    };
+
+    api->SectorsStatus = [=](SectorNumber id,
+                             bool) -> outcome::result<ApiSectorInfo> {
+      // TODO(ortyomka): [FIL-421] implement it
+      OUTCOME_TRY(sector_info, miner->getSealing()->getSectorInfo(id));
+      return ApiSectorInfo{.state = sector_info->state};
+    };
+
     api->StorageAttach = [=](const StorageInfo_ &storage_info,
                              const FsStat &stat) {
       return sector_index->storageAttach(storage_info, stat);

core/api/storage_miner/storage_api.hpp

@@ -59,6 +59,11 @@ namespace fc::api {
     PaddedPieceSize length;
   };
 
+  // TODO(ortyomka): [FIL-421] implement it
+  struct ApiSectorInfo {
+    mining::SealingState state = mining::SealingState::kStateUnknown;
+  };
+
   inline bool operator==(const PieceLocation &lhs, const PieceLocation &rhs) {
     return lhs.sector_number == rhs.sector_number && lhs.offset == rhs.offset
            && lhs.length == rhs.length;
@@ -106,6 +111,11 @@ namespace fc::api {
                void,
                const RetrievalAsk &)
 
+    API_METHOD(SectorsList, jwt::kReadPermission, std::vector<SectorNumber>)
+
+    API_METHOD(
+        SectorsStatus, jwt::kReadPermission, ApiSectorInfo, SectorNumber, bool)
+
     API_METHOD(StorageAttach,
                jwt::kAdminPermission,
                void,
@@ -233,6 +243,8 @@ namespace fc::api {
     f(a.MarketGetRetrievalAsk);
     f(a.MarketSetAsk);
     f(a.MarketSetRetrievalAsk);
+    f(a.SectorsList);
+    f(a.SectorsStatus);
     f(a.StorageAttach);
     f(a.StorageInfo);
     f(a.StorageReportHealth);

core/miner/main/main.cpp

@@ -32,6 +32,7 @@
 #include "markets/storage/chain_events/impl/chain_events_impl.hpp"
 #include "markets/storage/provider/impl/provider_impl.hpp"
 #include "miner/impl/miner_impl.hpp"
+#include "miner/main/type.hpp"
 #include "miner/mining.hpp"
 #include "miner/windowpost.hpp"
 #include "primitives/address/config.hpp"
@@ -61,13 +62,16 @@ namespace fc {
   using libp2p::multi::Multiaddress;
   using libp2p::peer::PeerId;
   using markets::storage::chain_events::ChainEventsImpl;
+  using primitives::StoredCounter;
   using primitives::address::Address;
   using primitives::jwt::kAllPermission;
   using primitives::sector::RegisteredSealProof;
   using storage::BufferMap;
   namespace uuids = boost::uuids;
 
   static const Buffer kActor{cbytes("actor")};
+  static const std::string kSectorCounterKey = "sector_counter";
+  constexpr size_t kApiThreadPoolSize = 4;
 
   auto log() {
     static common::Logger logger = common::createLogger("miner");
@@ -85,12 +89,42 @@ namespace fc {
 
     /** Path to presealed sectors */
     boost::optional<boost::filesystem::path> preseal_path;
+    boost::optional<boost::filesystem::path> preseal_meta_path;
 
     auto join(const std::string &path) const {
       return (repo_path / path).string();
     }
   };
 
+  outcome::result<void> migratePreSealMeta(
+      const std::string &path,
+      const Address &maddr,
+      const std::shared_ptr<storage::PersistentBufferMap> &ds) {
+    OUTCOME_TRY(file, common::readFile(path));
+    OUTCOME_TRY(j_file, codec::json::parse(gsl::make_span(file)));
+    OUTCOME_TRY(
+        psm, api::decode<std::map<std::string, miner::types::Miner>>(j_file));
+
+    const auto it_psm = psm.find(encodeToString(maddr));
+    if (it_psm == psm.end()) {
+      return ERROR_TEXT("Miner not found");
+    }
+    const auto &meta{it_psm->second};
+
+    StoredCounter sc(ds, kSectorCounterKey);
+    OUTCOME_TRY(max_sector, sc.getNumber());
+    for (const auto &elem : meta.sectors) {
+      // TODO(ortyomka): migrate sealing info
+
+      if (max_sector < elem.sector_id) {
+        max_sector = elem.sector_id;
+      }
+    }
+    OUTCOME_TRY(sc.setNumber(max_sector));
+
+    return outcome::success();
+  }
+
   outcome::result<Config> readConfig(int argc, char **argv) {
     namespace po = boost::program_options;
     Config config;
@@ -112,6 +146,9 @@ namespace fc {
     option("pre-sealed-sectors",
            po::value(&config.preseal_path),
            "Path to presealed sectors");
+    option("pre-sealed-metadata",
+           po::value(&config.preseal_meta_path),
+           "Path to presealed metadata");
     desc.add(configProfile());
     primitives::address::configCurrentNetwork(option);
 
@@ -301,9 +338,20 @@ namespace fc {
     IoThread sealing_thread;
 
     OUTCOME_TRY(setupMiner(config, *leveldb, host->getId()));
+    if (config.preseal_meta_path) {
+      log()->info("Importing pre-sealed sector metadata");
+      OUTCOME_TRY(migratePreSealMeta(config.preseal_meta_path.value().string(),
+                                     *config.actor,
+                                     prefixed("/metadata")));
+    }
 
     auto napi{std::make_shared<api::FullNodeApi>()};
-    api::rpc::Client wsc{*io};
+    IoThread io_thread;
+    std::vector<std::thread> pool;
+    for (size_t i{0}; i < kApiThreadPoolSize; ++i) {
+      pool.emplace_back(std::thread{[&] { io_thread.io->run(); }});
+    }
+    api::rpc::Client wsc{*io_thread.io};
     wsc.setup(*napi);
     OUTCOME_TRY(
         wsc.connect(config.node_api.first, "/rpc/v0", config.node_api.second));
@@ -352,32 +400,34 @@ namespace fc {
     auto remote_store{std::make_shared<sector_storage::stores::RemoteStoreImpl>(
         local_store, std::move(auth_headers))};
 
+    IoThread io_thread2;
+    OUTCOME_TRY(wscheduler,
+                sector_storage::SchedulerImpl::newScheduler(
+                    io_thread2.io, prefixed("scheduler_works/")));
+    IoThread io_thread3;
     OUTCOME_TRY(
-        wscheduler,
-        sector_storage::SchedulerImpl::newScheduler(
-            io, prefixed("scheduler_works/")));  // maybe use another io_context
-    OUTCOME_TRY(manager,
-                sector_storage::ManagerImpl::newManager(
-                    io, remote_store, wscheduler, {true, true, true, true}));
+        manager,
+        sector_storage::ManagerImpl::newManager(
+            io_thread3.io, remote_store, wscheduler, {true, true, true, true}));
 
     // TODO(ortyomka): make param
     mining::Config default_config{.max_wait_deals_sectors = 2,
                                   .max_sealing_sectors = 0,
                                   .max_sealing_sectors_for_deals = 0,
                                   .wait_deals_delay = std::chrono::hours(6)};
-    OUTCOME_TRY(
-        miner,
-        miner::MinerImpl::newMiner(napi,
-                                   *config.actor,
-                                   *config.worker,
-                                   std::make_shared<primitives::StoredCounter>(
-                                       leveldb, "sector_counter"),
-                                   prefixed("sealing_fsm/"),
-                                   manager,
-                                   scheduler,
-                                   sealing_thread.io,
-                                   default_config,
-                                   config.precommit_control));
+    OUTCOME_TRY(miner,
+                miner::MinerImpl::newMiner(
+                    napi,
+                    *config.actor,
+                    *config.worker,
+                    std::make_shared<primitives::StoredCounter>(
+                        prefixed("/metadata"), kSectorCounterKey),
+                    prefixed("sealing_fsm/"),
+                    manager,
+                    scheduler,
+                    sealing_thread.io,
+                    default_config,
+                    config.precommit_control));
     auto sealing{miner->getSealing()};
 
     OUTCOME_TRY(

core/miner/main/type.hpp

@@ -0,0 +1,43 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#include "primitives/address/address.hpp"
+#include "primitives/sector/sector.hpp"
+
+#include <libp2p/peer/peer_id.hpp>
+
+namespace fc::miner::types {
+  using libp2p::peer::PeerId;
+  using primitives::RegisteredSealProof;
+  using primitives::SectorNumber;
+  using primitives::SectorSize;
+  using primitives::TokenAmount;
+  using primitives::address::Address;
+  using vm::actor::builtin::types::market::DealProposal;
+
+  struct PreSealSector {
+    CID comm_r;
+    CID comm_d;
+    SectorNumber sector_id;
+    DealProposal deal;
+    RegisteredSealProof proof_type;
+  };
+
+  struct Miner {
+    Address id;
+    Address owner;
+    Address worker;
+    PeerId peer_id = codec::cbor::kDefaultT<PeerId>();
+
+    TokenAmount market_balance;
+    TokenAmount power_balance;
+
+    SectorSize sector_size{};
+
+    std::vector<PreSealSector> sectors;
+  };
+}  // namespace fc::miner::types