Merge pull request #2631 from Taraxa-project/release/v1.5.3

release v1.5.3

CMakeLists.txt

@@ -3,7 +3,7 @@ cmake_minimum_required(VERSION 3.20)
 # Set current version of the project
 set(TARAXA_MAJOR_VERSION 1)
 set(TARAXA_MINOR_VERSION 5)
-set(TARAXA_PATCH_VERSION 1)
+set(TARAXA_PATCH_VERSION 3)
 set(TARAXA_VERSION ${TARAXA_MAJOR_VERSION}.${TARAXA_MINOR_VERSION}.${TARAXA_PATCH_VERSION})
 
 # Any time a change in the network protocol is introduced this version should be increased

CMakeModules/ProjectJSONRPCCPP.cmake

@@ -4,10 +4,8 @@ set(include_path "${prefix}/include")
 
 ExternalProject_Add(jsonrpccpp
     PREFIX "${prefix}"
-    DOWNLOAD_NAME libjson-rpc-cpp-v1.4.1.tar.gz
-    DOWNLOAD_NO_PROGRESS TRUE
-    URL https://github.com/cinemast/libjson-rpc-cpp/archive/refs/tags/v1.4.1.tar.gz
-    URL_HASH SHA256=7a057e50d6203e4ea0a10ba5e4dbf344c48b177e5a3bf82e850eb3a783c11eb5
+    GIT_REPOSITORY https://github.com/MatusKysel/libjson-rpc-cpp.git
+    GIT_SHALLOW true
     CMAKE_ARGS
         -DCMAKE_BUILD_TYPE=Release
         -DCMAKE_INSTALL_PREFIX=<INSTALL_DIR>

libraries/cli/include/cli/config.hpp

@@ -59,6 +59,7 @@ class Config {
   static constexpr const char* ENABLE_TEST_RPC = "enable-test-rpc";
   static constexpr const char* ENABLE_DEBUG = "debug";
   static constexpr const char* MIGRATE_ONLY = "migrate-only";
+  static constexpr const char* FIX_TRX_PERIOD = "fix-transactions-period";
 
   std::string dirNameFromFile(const std::string& file);
 };

libraries/cli/src/config.cpp

@@ -48,6 +48,7 @@ Config::Config(int argc, const char* argv[]) {
   bool enable_test_rpc = false;
   bool enable_debug = false;
   bool migrate_only = false;
+  bool fix_trx_period = false;
 
   // Set node as default command
   command.push_back(NODE_COMMAND);
@@ -135,6 +136,8 @@ Config::Config(int argc, const char* argv[]) {
   node_command_options.add_options()(PRUNE_STATE_DB, bpo::bool_switch(&prune_state_db), "Prune state_db");
   node_command_options.add_options()(MIGRATE_ONLY, bpo::bool_switch(&migrate_only),
                                      "Only migrate DB, it will NOT run a node");
+  node_command_options.add_options()(FIX_TRX_PERIOD, bpo::bool_switch(&fix_trx_period),
+                                     "Fix transactions period field. This will take at least few hours");
 
   allowed_options.add(main_options);
 
@@ -270,6 +273,7 @@ Config::Config(int argc, const char* argv[]) {
     node_config_.db_config.prune_state_db = prune_state_db;
     node_config_.db_config.rebuild_db_period = rebuild_db_period;
     node_config_.db_config.migrate_only = migrate_only;
+    node_config_.db_config.fix_trx_period = fix_trx_period;
 
     node_config_.enable_test_rpc = enable_test_rpc;
     node_config_.enable_debug = enable_debug;

libraries/config/include/config/config.hpp

@@ -17,6 +17,7 @@ struct DBConfig {
   bool rebuild_db = false;
   bool prune_state_db = false;
   bool migrate_only = false;
+  bool fix_trx_period = false;
   PbftPeriod rebuild_db_period = 0;
 };
 

libraries/core_libs/network/graphql/src/ws_server.cpp

@@ -81,17 +81,7 @@ void GraphQlWsSession::triggerTestSubscribtion(unsigned int number) {
   res["params"] = params;
   auto response = util::to_string(res);
   ws_.text(ws_.got_text());
-  LOG(log_tr_) << "triggerTestSubscribtion: WS WRITE " << response.c_str();
-  auto executor = ws_.get_executor();
-  if (!executor) {
-    LOG(log_tr_) << "triggerTestSubscribtion: Executor missing - WS closed";
-    closed_ = true;
-    return;
-  }
-
-  LOG(log_tr_) << "***triggerTestSubscribtion: Before executor.post ";
-  boost::asio::post(executor, [this, response = std::move(response)]() mutable { writeImpl(std::move(response)); });
-  LOG(log_tr_) << "***triggerTestSubscribtion: After executors.post ";
+  writeAsync(std::move(response));
 }
 
 std::shared_ptr<WsSession> GraphQlWsServer::createSession(tcp::socket&& socket) {

libraries/core_libs/network/include/network/ws_server.hpp

@@ -33,7 +33,7 @@ class WsSession : public std::enable_shared_from_this<WsSession> {
  public:
   // Take ownership of the socket
   explicit WsSession(tcp::socket&& socket, addr_t node_addr, std::shared_ptr<WsServer> ws_server)
-      : ws_(std::move(socket)) {
+      : ws_(std::move(socket)), write_strand_(boost::asio::make_strand(ws_.get_executor())) {
     LOG_OBJECTS_CREATE("WS_SESSION");
     ws_server_ = ws_server;
   }
@@ -45,7 +45,6 @@ class WsSession : public std::enable_shared_from_this<WsSession> {
   void on_accept(beast::error_code ec);
   void do_read();
   void on_read(beast::error_code ec, std::size_t bytes_transferred);
-  void on_write_no_read(beast::error_code ec, std::size_t bytes_transferred);
 
   virtual std::string processRequest(const std::string_view& request) = 0;
 
@@ -56,15 +55,16 @@ class WsSession : public std::enable_shared_from_this<WsSession> {
   void newPendingTransaction(const trx_hash_t& trx_hash);
   bool is_closed() const { return closed_; }
   bool is_normal(const beast::error_code& ec) const;
+  void on_write(beast::error_code ec, std::size_t bytes_transferred);
   LOG_OBJECTS_DEFINE
 
  protected:
+  void processAsync();
+  void writeAsync(std::string&& message);
   void writeImpl(std::string&& message);
-  void write();
-  std::queue<std::string> queue_messages_;
   websocket::stream<beast::tcp_stream> ws_;
-  beast::flat_buffer buffer_;
-  std::string write_buffer_;
+  boost::asio::strand<boost::asio::any_io_executor> write_strand_;
+  beast::flat_buffer read_buffer_;
   int subscription_id_ = 0;
   int new_heads_subscription_ = 0;
   int new_dag_blocks_subscription_ = 0;

libraries/core_libs/network/rpc/Eth.jsonrpc.json

@@ -255,7 +255,8 @@
   {
     "name": "eth_estimateGas",
     "params": [
-      {}
+      {},
+      ""
     ],
     "order": [],
     "returns": ""

libraries/core_libs/network/rpc/EthClient.h

@@ -322,9 +322,10 @@ class EthClient : public jsonrpc::Client {
     else
       throw jsonrpc::JsonRpcException(jsonrpc::Errors::ERROR_CLIENT_INVALID_RESPONSE, result.toStyledString());
   }
-  std::string eth_estimateGas(const Json::Value& param1) throw(jsonrpc::JsonRpcException) {
+  std::string eth_estimateGas(const Json::Value& param1, const std::string& param2) throw(jsonrpc::JsonRpcException) {
     Json::Value p;
     p.append(param1);
+    p.append(param2);
     Json::Value result = this->CallMethod("eth_estimateGas", p);
     if (result.isString())
       return result.asString();

libraries/core_libs/network/rpc/EthFace.h

@@ -111,8 +111,8 @@ class EthFace : public ServerInterface<EthFace> {
                            &taraxa::net::EthFace::eth_sendRawTransactionI);
     this->bindAndAddMethod(jsonrpc::Procedure("eth_syncing", jsonrpc::PARAMS_BY_POSITION, jsonrpc::JSON_OBJECT, NULL),
                            &taraxa::net::EthFace::eth_syncingI);
-    this->bindAndAddMethod(jsonrpc::Procedure("eth_estimateGas", jsonrpc::PARAMS_BY_POSITION, jsonrpc::JSON_STRING,
-                                              "param1", jsonrpc::JSON_OBJECT, NULL),
+    this->bindAndAddMethod(jsonrpc::Procedure("eth_estimateGas", jsonrpc::PARAMS_BY_POSITION_WITH_OPTIONAL, jsonrpc::JSON_STRING,
+                                              "param1", jsonrpc::JSON_OBJECT, "param2", JSON_ANY, NULL),
                            &taraxa::net::EthFace::eth_estimateGasI);
     this->bindAndAddMethod(jsonrpc::Procedure("eth_chainId", jsonrpc::PARAMS_BY_POSITION, jsonrpc::JSON_OBJECT, NULL),
                            &taraxa::net::EthFace::eth_chainIdI);
@@ -223,7 +223,7 @@ class EthFace : public ServerInterface<EthFace> {
     response = this->eth_syncing();
   }
   inline virtual void eth_estimateGasI(const Json::Value &request, Json::Value &response) {
-    response = this->eth_estimateGas(request[0u]);
+    response = this->eth_estimateGas(request[0u], request[1u].asString());
   }
   inline virtual void eth_chainIdI(const Json::Value &request, Json::Value &response) {
     (void)request;
@@ -262,7 +262,7 @@ class EthFace : public ServerInterface<EthFace> {
   virtual Json::Value eth_getLogs(const Json::Value &param1) = 0;
   virtual std::string eth_sendRawTransaction(const std::string &param1) = 0;
   virtual Json::Value eth_syncing() = 0;
-  virtual std::string eth_estimateGas(const Json::Value &param1) = 0;
+  virtual std::string eth_estimateGas(const Json::Value &param1, const std::string &param2) = 0;
   virtual Json::Value eth_chainId() = 0;
 };
 

libraries/core_libs/network/rpc/eth/Eth.cpp

@@ -7,6 +7,7 @@
 #include <stdexcept>
 
 #include "LogFilter.hpp"
+#include "common/types.hpp"
 
 using namespace std;
 using namespace dev;
@@ -159,9 +160,14 @@ class EthImpl : public Eth, EthParams {
     return toJS(ret.code_retval);
   }
 
-  string eth_estimateGas(const Json::Value& _json) override {
+  string eth_estimateGas(const Json::Value& _json, const string& blockNumber) override {
     auto t = toTransactionSkeleton(_json);
-    auto blk_n = final_chain->last_block_number();
+    EthBlockNumber blk_n;
+    if (!blockNumber.empty()) {
+      blk_n = parse_blk_num(blockNumber);
+    } else {
+      blk_n = final_chain->last_block_number();
+    }
     prepare_transaction_for_call(t, blk_n);
 
     auto is_enough_gas = [&](gas_t gas) -> bool {

libraries/core_libs/network/src/ws_server.cpp

@@ -38,7 +38,7 @@ void WsSession::on_accept(beast::error_code ec) {
 
 void WsSession::do_read() {
   // Read a message into our buffer
-  ws_.async_read(buffer_, beast::bind_front_handler(&WsSession::on_read, shared_from_this()));
+  ws_.async_read(read_buffer_, beast::bind_front_handler(&WsSession::on_read, shared_from_this()));
 }
 
 void WsSession::on_read(beast::error_code ec, std::size_t bytes_transferred) {
@@ -51,42 +51,52 @@ void WsSession::on_read(beast::error_code ec, std::size_t bytes_transferred) {
     return close(is_normal(ec));
   }
 
-  LOG(log_tr_) << "WS READ " << (static_cast<char *>(buffer_.data().data()));
+  LOG(log_tr_) << "WS READ " << (static_cast<char *>(read_buffer_.data().data()));
 
-  const std::string_view str_view(static_cast<const char *>(buffer_.data().data()), buffer_.size());
-  const auto response = processRequest(str_view);
-
-  // Clear the buffer
-  buffer_.consume(buffer_.size());
+  processAsync();
+  // Do another read
+  do_read();
+}
 
+void WsSession::processAsync() {
+  std::string request(static_cast<char *>(read_buffer_.data().data()), read_buffer_.size());
+  read_buffer_.consume(read_buffer_.size());
+  LOG(log_tr_) << "processAsync " << request;
   auto executor = ws_.get_executor();
   if (!executor) {
     LOG(log_tr_) << "Executor missing - WS closed";
-    return close(false);
+    closed_ = true;
+    return;
   }
 
-  boost::asio::post(executor, [this, response = std::move(response)]() mutable { writeImpl(std::move(response)); });
-  // Do another read
-  do_read();
+  LOG(log_tr_) << "Before executor.post ";
+  boost::asio::post(executor, [this, request = std::move(request)]() mutable { writeAsync(processRequest(request)); });
+  LOG(log_tr_) << "After executor.post ";
 }
 
-void WsSession::on_write_no_read(beast::error_code ec, std::size_t bytes_transferred) {
-  LOG(log_tr_) << "WS ASYNC WRITE COMPLETE"
-               << " " << &ws_;
-  // Pop on successful write
-  queue_messages_.pop();
-  if (is_closed()) return;
-
-  // For any error close the connection
-  if (ec) {
-    LOG(log_nf_) << "WS closed in on_write " << ec;
-    return close(is_normal(ec));
+void WsSession::writeAsync(std::string &&message) {
+  LOG(log_tr_) << "WS WRITE " << message.c_str();
+  auto executor = ws_.get_executor();
+  if (!executor) {
+    LOG(log_tr_) << "Executor missing - WS closed";
+    closed_ = true;
+    return;
   }
 
-  boost::ignore_unused(bytes_transferred);
-  if (queue_messages_.size() > 0) {
-    write();
+  LOG(log_tr_) << "Before executor.post ";
+  boost::asio::post(write_strand_, [this, message = std::move(message)]() mutable { writeImpl(std::move(message)); });
+  LOG(log_tr_) << "After executors.post ";
+}
+
+void WsSession::writeImpl(std::string &&message) {
+  ws_.text(true);  // as we are using text msg here
+  try {
+    ws_.write(boost::asio::buffer(message));
+  } catch (const boost::system::system_error &e) {
+    LOG(log_nf_) << "WS closed in on_write " << e.what();
+    return close(is_normal(e.code()));
   }
+  LOG(log_tr_) << "WS WRITE COMPLETE " << &ws_;
 }
 
 void WsSession::newEthBlock(const ::taraxa::final_chain::BlockHeader &payload, const TransactionHashes &trx_hashes) {
@@ -100,32 +110,9 @@ void WsSession::newEthBlock(const ::taraxa::final_chain::BlockHeader &payload, c
     res["params"] = params;
     auto response = util::to_string(res);
     LOG(log_tr_) << "WS WRITE " << response.c_str();
-    auto executor = ws_.get_executor();
-    if (!executor) {
-      LOG(log_tr_) << "Executor missing - WS closed";
-      return close(false);
-    }
-    boost::asio::post(executor, [this, response = std::move(response)]() mutable { writeImpl(std::move(response)); });
+    writeAsync(std::move(response));
   }
 }
-
-void WsSession::write() {
-  write_buffer_ = std::move(queue_messages_.front());
-  ws_.text(true);  // as we are using text msg here
-  LOG(log_tr_) << "WS ASYNC WRITE " << write_buffer_.c_str() << " " << &ws_;
-  ws_.async_write(boost::asio::buffer(write_buffer_),
-                  beast::bind_front_handler(&WsSession::on_write_no_read, shared_from_this()));
-}
-
-void WsSession::writeImpl(std::string &&message) {
-  queue_messages_.push(std::move(message));
-  if (queue_messages_.size() > 1) {
-    // outstanding async_write
-    return;
-  }
-  write();
-}
-
 void WsSession::newDagBlock(DagBlock const &blk) {
   if (new_dag_blocks_subscription_) {
     Json::Value res, params;
@@ -140,7 +127,7 @@ void WsSession::newDagBlock(DagBlock const &blk) {
       LOG(log_tr_) << "Executor missing - WS closed";
       return close(false);
     }
-    boost::asio::post(executor, [this, response = std::move(response)]() mutable { writeImpl(std::move(response)); });
+    writeAsync(std::move(response));
   }
 }
 
@@ -155,12 +142,7 @@ void WsSession::newDagBlockFinalized(blk_hash_t const &blk, uint64_t period) {
     params["subscription"] = dev::toJS(new_dag_block_finalized_subscription_);
     res["params"] = params;
     auto response = util::to_string(res);
-    auto executor = ws_.get_executor();
-    if (!executor) {
-      LOG(log_tr_) << "Executor missing - WS closed";
-      return close(false);
-    }
-    boost::asio::post(executor, [this, response = std::move(response)]() mutable { writeImpl(std::move(response)); });
+    writeAsync(std::move(response));
   }
 }
 
@@ -174,12 +156,7 @@ void WsSession::newPbftBlockExecuted(Json::Value const &payload) {
     params["subscription"] = dev::toJS(new_pbft_block_executed_subscription_);
     res["params"] = params;
     auto response = util::to_string(res);
-    auto executor = ws_.get_executor();
-    if (!executor) {
-      LOG(log_tr_) << "Executor missing - WS closed";
-      return close(false);
-    }
-    boost::asio::post(executor, [this, response = std::move(response)]() mutable { writeImpl(std::move(response)); });
+    writeAsync(std::move(response));
   }
 }
 
@@ -192,12 +169,7 @@ void WsSession::newPendingTransaction(trx_hash_t const &trx_hash) {
     params["subscription"] = dev::toJS(new_transactions_subscription_);
     res["params"] = params;
     auto response = util::to_string(res);
-    auto executor = ws_.get_executor();
-    if (!executor) {
-      LOG(log_tr_) << "Executor missing - WS closed";
-      return close(false);
-    }
-    boost::asio::post(executor, [this, response = std::move(response)]() mutable { writeImpl(std::move(response)); });
+    writeAsync(std::move(response));
   }
 }
 
@@ -266,8 +238,7 @@ WsServer::~WsServer() {
 
 void WsServer::do_accept() {
   // The new connection gets its own strand
-  acceptor_.async_accept(boost::asio::make_strand(ioc_),
-                         beast::bind_front_handler(&WsServer::on_accept, shared_from_this()));
+  acceptor_.async_accept(ioc_, beast::bind_front_handler(&WsServer::on_accept, shared_from_this()));
 }
 
 void WsServer::on_accept(beast::error_code ec, tcp::socket socket) {

libraries/core_libs/node/src/node.cpp

@@ -28,6 +28,7 @@
 #include "pbft/pbft_manager.hpp"
 #include "slashing_manager/slashing_manager.hpp"
 #include "storage/migration/migration_manager.hpp"
+#include "storage/migration/transaction_period.hpp"
 #include "transaction/gas_pricer.hpp"
 #include "transaction/transaction_manager.hpp"
 
@@ -87,7 +88,11 @@ void FullNode::init() {
 
     db_->updateDbVersions();
 
-    storage::migration::Manager(db_).applyAll();
+    auto migration_manager = storage::migration::Manager(db_);
+    migration_manager.applyAll();
+    if (conf_.db_config.fix_trx_period) {
+      migration_manager.applyTransactionPeriod();
+    }
 
     if (db_->getDagBlocksCount() == 0) {
       db_->setGenesisHash(conf_.genesis.genesisHash());