Feature/republish (#490)

mpool republish pending messages

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

Author: Alexey-N-Chernyshov

core/api/full_node/make.cpp

@@ -333,7 +333,7 @@ namespace fc::api {
             auto tag,
             auto epoch,
             auto &entropy) -> outcome::result<Randomness> {
-      std::shared_lock ts_lock{*env_context.ts_branches_mutex};
+      std::unique_lock ts_lock{*env_context.ts_branches_mutex};
       OUTCOME_TRY(ts_branch, TsBranch::make(ts_load, tipset_key, ts_main));
       return env_context.randomness->getRandomnessFromBeacon(
           ts_branch, tag, epoch, entropy);
@@ -343,7 +343,7 @@ namespace fc::api {
             auto tag,
             auto epoch,
             auto &entropy) -> outcome::result<Randomness> {
-      std::shared_lock ts_lock{*env_context.ts_branches_mutex};
+      std::unique_lock ts_lock{*env_context.ts_branches_mutex};
       OUTCOME_TRY(ts_branch, TsBranch::make(ts_load, tipset_key, ts_main));
       return env_context.randomness->getRandomnessFromTickets(
           ts_branch, tag, epoch, entropy);
@@ -353,7 +353,7 @@ namespace fc::api {
     };
     api->ChainGetTipSetByHeight =
         [=](auto height, auto tipset_key) -> outcome::result<TipsetCPtr> {
-      std::shared_lock ts_lock{*env_context.ts_branches_mutex};
+      std::unique_lock ts_lock{*env_context.ts_branches_mutex};
       if (tipset_key.cids().empty()) {
         tipset_key = chain_store->heaviestTipset()->key;
       }
@@ -589,7 +589,7 @@ namespace fc::api {
           OUTCOME_CB(auto context, tipsetContext(tipset_key, true));
           MiningBaseInfo info;
 
-          std::shared_lock ts_lock{*env_context.ts_branches_mutex};
+          std::unique_lock ts_lock{*env_context.ts_branches_mutex};
           OUTCOME_CB(auto ts_branch,
                      TsBranch::make(ts_load, tipset_key, ts_main));
           OUTCOME_CB(auto it, find(ts_branch, context.tipset->height()));
@@ -667,7 +667,8 @@ namespace fc::api {
       OUTCOME_TRY(signed_message,
                   vm::message::MessageSignerImpl{key_store}.sign(message.from,
                                                                  message));
-      OUTCOME_TRY(mpool->add(signed_message));
+      OUTCOME_TRY(mpool->addLocal(signed_message));
+      mpool->publish(signed_message);
       spdlog::info("MpoolPushMessage {}", signed_message.getCid());
       return std::move(signed_message);
     };
@@ -699,7 +700,7 @@ namespace fc::api {
                          auto &tipset_key) -> outcome::result<InvocResult> {
       OUTCOME_TRY(context, tipsetContext(tipset_key));
 
-      std::shared_lock ts_lock{*env_context.ts_branches_mutex};
+      std::unique_lock ts_lock{*env_context.ts_branches_mutex};
       OUTCOME_TRY(ts_branch, TsBranch::make(ts_load, tipset_key, ts_main));
       ts_lock.unlock();
 

core/codec/cbor/fwd.hpp

@@ -10,7 +10,10 @@
 
 namespace fc::codec::cbor {
   template <typename T>
+  // NOLINTNEXTLINE(readability-redundant-declaration)
   outcome::result<Bytes> encode(const T &arg);
+
   template <typename T>
+  // NOLINTNEXTLINE(readability-redundant-declaration)
   outcome::result<T> decode(BytesIn input);
 }  // namespace fc::codec::cbor

core/common/outcome_fmt.hpp

@@ -19,7 +19,7 @@ struct fmt::formatter<std::error_code, char, void> {
 
   template <typename ParseContext>
   constexpr auto parse(ParseContext &ctx) {
-    auto it{ctx.begin()};
+    const auto *it{ctx.begin()};
     if (it != ctx.end() && *it == '#') {
       alt = true;
       ++it;

core/const.hpp

@@ -12,6 +12,7 @@ namespace fc {
   using primitives::ChainEpoch;
   using primitives::EpochDuration;
   using primitives::StoragePower;
+  using primitives::TokenAmount;
 
   constexpr int64_t kSecondsInHour{60 * 60};
   constexpr int64_t kSecondsInDay{24 * kSecondsInHour};
@@ -38,7 +39,7 @@ namespace fc {
   constexpr uint64_t kFilecoinPrecision{1000000000000000000};
   constexpr auto kGasLimitOverestimation{1.25};
   constexpr auto kMessageConfidence{5};
-  constexpr auto kMinimumBaseFee{100};
+  const TokenAmount kMinimumBaseFee{100};
   constexpr auto kPackingEfficiencyDenom{5};
   constexpr auto kPackingEfficiencyNum{4};
 

core/node/main/builder.cpp

@@ -261,7 +261,35 @@ namespace fc::node {
           timerLoop(scheduler, tick, cb);
         },
         tick);
-  };
+  }
+
+  /**
+   * Creates and initializes message pool and sets timer for republishing.
+   * @param node_objects
+   * @return
+   */
+  void createMessagePool(const Config &config, NodeObjects &o) {
+    o.mpool = storage::mpool::MessagePool::create(o.env_context,
+                                                  o.ts_main,
+                                                  config.mpool_bls_cache_size,
+                                                  o.chain_store,
+                                                  o.pubsub_gate);
+    // Republish pending messages
+    // Delay from lotus
+    // https://github.com/filecoin-project/lotus/blob/d9100981ada8b3186d906a4f4140b83a819d2299/chain/messagepool/messagepool.go#L58
+    const auto republishTimeout{std::chrono::seconds(10 * kEpochDurationSeconds
+                                                     + kPropagationDelaySecs)};
+    timerLoop(o.scheduler, republishTimeout, [mpool{o.mpool}] {
+      const auto res = mpool->republishPendingMessages();
+      if (!res) {
+        log()->error("Mpool republish error: {:#}", res.error());
+      }
+    });
+    // batch message publishing with delay kRepublishBatchDelay
+    timerLoop(o.scheduler,
+              storage::mpool::kRepublishBatchDelay,
+              [mpool{o.mpool}] { mpool->publishFromQueue(); });
+  }
 
   /**
    * Creates and intialises Storage Market Client
@@ -306,6 +334,7 @@ namespace fc::node {
     return outcome::success();
   }
 
+  // NOLINTNEXTLINE(readability-function-cognitive-complexity)
   outcome::result<NodeObjects> createNodeObjects(Config &config) {
     NodeObjects o;
 
@@ -524,8 +553,7 @@ namespace fc::node {
 
     log()->debug("Creating API...");
 
-    o.mpool = storage::mpool::MessagePool::create(
-        o.env_context, o.ts_main, config.mpool_bls_cache_size, o.chain_store);
+    createMessagePool(config, o);
 
     auto msg_waiter = storage::blockchain::MsgWaiter::create(
         o.ts_load, o.ipld, o.io_context, o.chain_store);

core/node/main/main.cpp

@@ -277,14 +277,6 @@ namespace fc {
                       res.error());
       }
     })};
-    o.api->MpoolPushMessage = [&, impl{o.api->MpoolPushMessage}](auto &arg1,
-                                                                 auto &arg2) {
-      auto res{impl(arg1, arg2)};
-      if (res) {
-        o.pubsub_gate->publish(res.value());
-      }
-      return res;
-    };
 
     Metrics metrics{o, start_time};
 

core/storage/buffer_map.hpp

@@ -21,7 +21,7 @@
 
 namespace fc::storage {
 
-  // TODO: BytesIn as key or remove all of them
+  // TODO(ortyomka): change `K` to `cow_t<K>` in interfaces
   using BufferMap = face::GenericMap<Bytes, Bytes>;
 
   using BufferBatch = face::WriteBatch<Bytes, Bytes>;

core/storage/mpool/CMakeLists.txt

@@ -6,4 +6,5 @@ add_library(mpool
     )
 target_link_libraries(mpool
     message
+    sync
     )