Commit batcher

core/miner/address_selector.hpp

@@ -18,14 +18,15 @@ namespace fc::mining {
 
   /**
    * SelectAddress takes the maximal possible transaction fee from configs and
-   * chooses one of control addresses with minimal balance that is more than good
-   * funds to make miner work as long as possible. If no suitble control address
-   * were found, function returns worker address.
+   * chooses one of control addresses with minimal balance that is more than
+   * good funds to make miner work as long as possible. If no suitble control
+   * address were found, function returns worker address.
    */
   inline outcome::result<Address> SelectAddress(
       const MinerInfo &miner_info,
       const TokenAmount &good_funds,
-      const std::shared_ptr<FullNodeApi> &api) {
+      const std::shared_ptr<FullNodeApi>
+          &api) {  // TODO update from lotus (Markuuu-s)
     TokenAmount finder_balance;
     auto finder = miner_info.control.end();
     for (auto address = miner_info.control.begin();

core/miner/storage_fsm/CMakeLists.txt

@@ -38,6 +38,7 @@ target_link_libraries(events
 
 add_library(batcher
         impl/precommit_batcher_impl.cpp
+        impl/commit_batcher_impl.cpp
         )
 target_link_libraries(batcher
         api

core/miner/storage_fsm/commit_batcher.hpp

@@ -0,0 +1,37 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+#include "api/full_node/node_api.hpp"
+#include "miner/storage_fsm/types.hpp"
+#include "primitives/sector/sector.hpp"
+
+namespace fc::mining {
+  using CommitCallback = std::function<void(const outcome::result<CID> &)>;
+  using api::FullNodeApi;
+  using primitives::sector::AggregateSealVerifyInfo;
+  using primitives::sector::RegisteredSealProof;
+  using sector_storage::Proof;
+  using types::SectorInfo;
+
+  struct AggregateInput {
+    Proof proof;
+    AggregateSealVerifyInfo info;
+    RegisteredSealProof spt;
+  };
+
+  class CommitBatcher {
+   public:
+    virtual ~CommitBatcher() = default;
+
+    virtual outcome::result<void> addCommit(
+        const SectorInfo &sector_info,
+        const AggregateInput &aggregate_input,
+        const CommitCallback &callBack) = 0;
+
+    virtual void forceSend() = 0;
+  };
+
+}  // namespace fc::mining

core/miner/storage_fsm/impl/commit_batcher_impl.cpp

@@ -0,0 +1,246 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+#include "commit_batcher_impl.hpp"
+#include <iostream>
+#include <utility>
+#include "vm/actor/builtin/v5/miner/miner_actor.hpp"
+#include "vm/actor/builtin/v6/monies.hpp"
+
+#include <string>
+
+namespace fc::mining {
+  using api::kPushNoSpec;
+  using fc::BytesIn;
+  using fc::proofs::ProofEngine;
+  using primitives::ActorId;
+  using primitives::go::bigdiv;
+  using primitives::sector::AggregateSealVerifyInfo;
+  using primitives::sector::AggregateSealVerifyProofAndInfos;
+  using primitives::tipset::TipsetCPtr;
+  using vm::actor::MethodParams;
+  using vm::actor::builtin::types::miner::kChainFinality;
+  using vm::actor::builtin::types::miner::SectorPreCommitOnChainInfo;
+  using vm::actor::builtin::v5::miner::ProveCommitAggregate;
+  using vm::actor::builtin::v6::miner::AggregateProveCommitNetworkFee;
+
+  CommitBatcherImpl::CommitBatcherImpl(const std::chrono::seconds &max_time,
+                                       std::shared_ptr<FullNodeApi> api,
+                                       Address miner_address,
+                                       std::shared_ptr<Scheduler> scheduler,
+                                       AddressSelector address_selector,
+                                       std::shared_ptr<FeeConfig> fee_config,
+                                       const size_t &max_size_callback,
+                                       std::shared_ptr<ProofEngine> proof)
+      : scheduler_(std::move(scheduler)),
+        max_delay_(max_time),
+        closest_cutoff_(max_time),
+        max_size_callback_(max_size_callback),
+        api_(std::move(api)),
+        miner_address_(std::move(miner_address)),
+        fee_config_(std::move(fee_config)),
+        proof_(std::move(proof)),
+        address_selector_(std::move(address_selector)) {
+    cutoff_start_ = std::chrono::system_clock::now();
+    reschedule(max_delay_);
+  }
+
+  outcome::result<void> CommitBatcherImpl::addCommit(
+      const SectorInfo &sector_info,
+      const AggregateInput &aggregate_input,
+      const CommitCallback &callback) {
+    std::unique_lock<std::mutex> locker(mutex_storage_);
+
+    const SectorNumber &sector_number = sector_info.sector_number;
+    OUTCOME_TRY(head, api_->ChainHead());
+
+    pair_storage_[sector_number] = PairStorage{aggregate_input, callback};
+
+    if (pair_storage_.size() >= max_size_callback_) {
+      locker.unlock();
+      forceSend();
+      return outcome::success();
+    }
+
+    setCommitCutoff(head->epoch(), sector_info);
+    return outcome::success();
+  }
+
+  void CommitBatcherImpl::forceSend() {
+    MapPairStorage pair_storage_for_send_;
+
+    std::unique_lock<std::mutex> locker(mutex_storage_);
+    pair_storage_for_send_ = std::move(pair_storage_);
+    locker.unlock();
+
+    const auto maybe_result = sendBatch(pair_storage_for_send_);
+
+    for (auto &[key, pair_storage] : pair_storage_for_send_) {
+      pair_storage.commit_callback(maybe_result);
+    }
+
+    cutoff_start_ = std::chrono::system_clock::now();
+    closest_cutoff_ = max_delay_;
+    reschedule(max_delay_);
+  }
+
+  void CommitBatcherImpl::reschedule(std::chrono::milliseconds time) {
+    handle_ = scheduler_->scheduleWithHandle(
+        [&]() {
+          MapPairStorage pair_storage_for_send_;
+
+          std::unique_lock<std::mutex> locker(mutex_storage_);
+          pair_storage_for_send_ = std::move(pair_storage_);
+          locker.unlock();
+
+          const auto maybe_result = sendBatch(pair_storage_for_send_);
+
+          for (auto &[key, pair_storage] : pair_storage_for_send_) {
+            pair_storage.commit_callback(maybe_result);
+          }
+
+          cutoff_start_ = std::chrono::system_clock::now();
+          closest_cutoff_ = max_delay_;
+
+          handle_.reschedule(max_delay_).value();
+        },
+        time);
+  }
+
+  outcome::result<CID> CommitBatcherImpl::sendBatch(
+      const MapPairStorage &pair_storage_for_send) {
+    if (pair_storage_for_send.empty()) {
+      cutoff_start_ = std::chrono::system_clock::now();
+      return ERROR_TEXT("Empty Batcher");
+    }
+    OUTCOME_TRY(head, api_->ChainHead());
+
+    const size_t total = pair_storage_for_send.size();
+
+    ProveCommitAggregate::Params params;
+
+    std::vector<Proof> proofs;
+    proofs.reserve(total);
+
+    std::vector<AggregateSealVerifyInfo> infos;
+    infos.reserve(total);
+
+    BigInt collateral = 0;
+
+    for (const auto &[sector_number, pair_storage] : pair_storage_for_send) {
+      OUTCOME_TRY(sector_collateral,
+                  getSectorCollateral(sector_number, head->key));
+      collateral = collateral + sector_collateral;
+
+      params.sectors.insert(sector_number);
+      infos.push_back(pair_storage.aggregate_input.info);
+    }
+
+    for (const auto &[sector_number, pair_storage] : pair_storage_for_send) {
+      proofs.push_back(pair_storage.aggregate_input.proof);
+    }
+
+    const ActorId mid = miner_address_.getId();
+    // TODO maybe long (AggregateSealProofs)
+    AggregateSealVerifyProofAndInfos aggregate_seal{
+        .miner = mid,
+        .seal_proof =
+            pair_storage_for_send.at(infos[0].number).aggregate_input.spt,
+        .aggregate_proof = arp_,
+        .infos = infos};
+
+    std::vector<BytesIn> proofsSpan;
+    proofsSpan.reserve(proofs.size());
+
+    for (const Proof &proof : proofs) {
+      proofsSpan.push_back(gsl::make_span(proof));
+    }
+
+    OUTCOME_TRY(proof_->aggregateSealProofs(aggregate_seal, proofsSpan));
+
+    params.proof = aggregate_seal.proof;
+    OUTCOME_TRY(encode, codec::cbor::encode(params));
+    OUTCOME_TRY(miner_info, api_->StateMinerInfo(miner_address_, head->key));
+
+    const TokenAmount max_fee =
+        fee_config_->max_commit_batch_gas_fee.FeeForSector(proofs.size());
+
+    OUTCOME_TRY(tipset, api_->ChainGetTipSet(head->key));
+    const BigInt base_fee = tipset->blks[0].parent_base_fee;
+
+    TokenAmount agg_fee_raw =
+        AggregateProveCommitNetworkFee(infos.size(), base_fee); // TODO change to aggregateNetworkFee
+
+    TokenAmount agg_fee = bigdiv(agg_fee_raw * agg_fee_num_, agg_fee_den_);
+    TokenAmount need_funds = collateral + agg_fee;
+    TokenAmount good_funds = max_fee + need_funds;
+
+    OUTCOME_TRY(address,
+                address_selector_(miner_info, good_funds, need_funds, api_));
+    OUTCOME_TRY(signed_messege,
+                api_->MpoolPushMessage(
+                    vm::message::UnsignedMessage(miner_address_,
+                                                 address,
+                                                 0,
+                                                 need_funds,
+                                                 max_fee,
+                                                 {},
+                                                 ProveCommitAggregate::Number,
+                                                 MethodParams{encode}),
+                    kPushNoSpec));
+
+    cutoff_start_ = std::chrono::system_clock::now();
+    return signed_messege.getCid();
+  }
+
+  void CommitBatcherImpl::setCommitCutoff(const ChainEpoch &current_epoch,
+                                          const SectorInfo &sector_info) {
+    ChainEpoch cutoff_epoch =
+        sector_info.ticket_epoch
+        + static_cast<int64_t>(kEpochsInDay + kChainFinality);
+    ChainEpoch start_epoch{};
+    for (const auto &piece : sector_info.pieces) {
+      if (!piece.deal_info) {
+        continue;
+      }
+      start_epoch = piece.deal_info->deal_schedule.start_epoch;
+      if (start_epoch < cutoff_epoch) {
+        cutoff_epoch = start_epoch;
+      }
+    }
+    if (cutoff_epoch <= current_epoch) {
+      forceSend();
+    } else {
+      const auto temp_cutoff = std::chrono::milliseconds(
+          (cutoff_epoch - current_epoch) * kEpochDurationSeconds);
+      if ((closest_cutoff_
+               - std::chrono::duration_cast<std::chrono::milliseconds>(
+                   std::chrono::system_clock::now() - cutoff_start_)
+           > temp_cutoff)) {
+        cutoff_start_ = std::chrono::system_clock::now();
+        reschedule(temp_cutoff);
+        closest_cutoff_ = temp_cutoff;
+      }
+    }
+  }
+
+  outcome::result<TokenAmount> CommitBatcherImpl::getSectorCollateral(
+      const SectorNumber &sector_number, const TipsetKey &tip_set_key) {
+    OUTCOME_TRY(pci,
+                api_->StateSectorPreCommitInfo(
+                    miner_address_, sector_number, tip_set_key));
+
+    OUTCOME_TRY(collateral,
+                api_->StateMinerInitialPledgeCollateral(
+                    miner_address_, pci.info, tip_set_key));
+
+    collateral = collateral + pci.precommit_deposit;
+    collateral = std::max(BigInt(0), collateral);
+
+    return collateral;
+  }
+
+}  // namespace fc::mining

core/miner/storage_fsm/impl/commit_batcher_impl.hpp

@@ -0,0 +1,83 @@
+/**
+ * Copyright Soramitsu Co., Ltd. All Rights Reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+#include <libp2p/basic/scheduler.hpp>
+#include "miner/storage_fsm/commit_batcher.hpp"
+
+namespace fc::mining {
+  using api::MinerInfo;
+  using libp2p::basic::Scheduler;
+  using primitives::BigInt;
+  using primitives::SectorNumber;
+  using primitives::address::Address;
+  using primitives::sector::RegisteredAggregationProof;
+  using primitives::tipset::Tipset;
+  using primitives::tipset::TipsetKey;
+  using proofs::ProofEngine;
+  using types::FeeConfig;
+
+  using AddressSelector = std::function<outcome::result<Address>(
+      const MinerInfo &miner_info,
+      const TokenAmount &good_funds,
+      const TokenAmount &need_funds,
+      const std::shared_ptr<FullNodeApi> &api)>;
+
+  class CommitBatcherImpl : public CommitBatcher {
+   public:
+    struct PairStorage {
+      AggregateInput aggregate_input;
+      CommitCallback commit_callback;
+    };
+    using MapPairStorage = std::map<SectorNumber, PairStorage>;
+
+    CommitBatcherImpl(const std::chrono::seconds &max_time,
+                      std::shared_ptr<FullNodeApi> api,
+                      Address miner_address,
+                      std::shared_ptr<Scheduler> scheduler,
+                      AddressSelector address_selector,
+                      std::shared_ptr<FeeConfig> fee_config,
+                      const size_t &max_size_callback,
+                      std::shared_ptr<ProofEngine> proof);
+
+    outcome::result<void> addCommit(const SectorInfo &sector_info,
+                                    const AggregateInput &aggregate_input,
+                                    const CommitCallback &callBack) override;
+
+    void forceSend() override;
+
+    void setCommitCutoff(const ChainEpoch &current_epoch,
+                         const SectorInfo &sector_info);
+
+   private:
+    std::shared_ptr<Scheduler> scheduler_;
+    Scheduler::Handle handle_;
+    std::chrono::milliseconds max_delay_;
+    std::chrono::milliseconds closest_cutoff_;
+    std::chrono::system_clock::time_point cutoff_start_;
+    size_t max_size_callback_;
+    MapPairStorage pair_storage_;
+    std::shared_ptr<FullNodeApi> api_;
+    Address miner_address_;
+    std::shared_ptr<FeeConfig> fee_config_;
+    std::shared_ptr<ProofEngine> proof_;
+    std::mutex mutex_storage_;
+    AddressSelector address_selector_;
+    common::Logger logger_;
+
+    const BigInt agg_fee_num_ = BigInt(110);
+    const BigInt agg_fee_den_ = BigInt(100);
+    const RegisteredAggregationProof arp_ = RegisteredAggregationProof(0);
+
+    // TODO (Markuu-s) Add processIndividually
+    void reschedule(std::chrono::milliseconds time);
+
+    outcome::result<CID> sendBatch(const MapPairStorage &pair_storage_for_send);
+
+    outcome::result<TokenAmount> getSectorCollateral(
+        const SectorNumber &sector_number, const TipsetKey &tip_set_key);
+  };
+
+}  // namespace fc::mining