GH-109 Add /v1/chain/get_unapplied_transactions endpoint to producer_…

…api_plugin

plugins/producer_api_plugin/producer_api_plugin.cpp

@@ -29,10 +29,11 @@ struct async_result_visitor : public fc::visitor<fc::variant> {
 
 #define CALL_WITH_400(api_name, api_handle, call_name, INVOKE, http_response_code) \
 {std::string("/v1/" #api_name "/" #call_name), \
-   [&api_handle](string, string body, url_response_callback cb) mutable { \
+   [&api_handle, http_max_response_time](string, string body, url_response_callback cb) mutable { \
           try { \
+             auto deadline = fc::time_point::now() + http_max_response_time; \
              INVOKE \
-             cb(http_response_code, fc::time_point::maximum(), fc::variant(result)); \
+             cb(http_response_code, deadline, fc::variant(result)); \
           } catch (...) { \
              http_plugin::handle_exception(#api_name, #call_name, body, cb); \
           } \
@@ -65,6 +66,10 @@ struct async_result_visitor : public fc::visitor<fc::variant> {
      auto params = parse_params<in_param, http_params_types::possible_no_params>(body);\
      auto result = api_handle.call_name(std::move(params));
 
+#define INVOKE_R_R_D(api_handle, call_name, in_param) \
+     auto params = parse_params<in_param, http_params_types::possible_no_params>(body);\
+     auto result = api_handle.call_name(std::move(params), deadline);
+
 #define INVOKE_R_V(api_handle, call_name) \
      body = parse_params<std::string, http_params_types::no_params>(body); \
      auto result = api_handle.call_name();
@@ -87,6 +92,8 @@ void producer_api_plugin::plugin_startup() {
    ilog("starting producer_api_plugin");
    // lifetime of plugin is lifetime of application
    auto& producer = app().get_plugin<producer_plugin>();
+   auto& http = app().get_plugin<http_plugin>();
+   fc::microseconds http_max_response_time = http.get_max_response_time();
 
    app().get_plugin<http_plugin>().add_api({
        CALL_WITH_400(producer, producer, pause,
@@ -122,6 +129,8 @@ void producer_api_plugin::plugin_startup() {
                                  producer_plugin::get_supported_protocol_features_params), 201),
        CALL_WITH_400(producer, producer, get_account_ram_corrections,
             INVOKE_R_R(producer, get_account_ram_corrections, producer_plugin::get_account_ram_corrections_params), 201),
+       CALL_WITH_400(producer, producer, get_unapplied_transactions,
+                     INVOKE_R_R_D(producer, get_unapplied_transactions, producer_plugin::get_unapplied_transactions_params), 200),
    }, appbase::priority::medium_high);
 }
 

plugins/producer_plugin/include/eosio/producer_plugin/producer_plugin.hpp

@@ -114,7 +114,35 @@ class producer_plugin : public appbase::plugin<producer_plugin> {
 
    get_account_ram_corrections_result  get_account_ram_corrections( const get_account_ram_corrections_params& params ) const;
 
-    void log_failed_transaction(const transaction_id_type& trx_id, const chain::packed_transaction_ptr& packed_trx_ptr, const char* reason) const;
+   struct get_unapplied_transactions_params {
+      string      lower_bound;  /// transaction id
+      std::optional<uint32_t>    limit = 100;
+      std::optional<uint32_t>    time_limit_ms; // defaults to 10ms
+   };
+
+   struct unapplied_trx {
+      transaction_id_type       trx_id;
+      fc::time_point_sec        expiration;
+      string                    trx_type; // eosio::chain::trx_enum_type values or "read_only"
+      account_name              first_auth;
+      account_name              first_receiver;
+      action_name               first_action;
+      uint16_t                  total_actions = 0;
+      uint32_t                  billed_cpu_time_us = 0;
+      size_t                    size = 0;
+   };
+
+   struct get_unapplied_transactions_result {
+      size_t                     size = 0;
+      size_t                     incoming_size = 0;
+      std::vector<unapplied_trx> trxs;
+      string                     more; ///< fill lower_bound with trx id to fetch next set of transactions
+   };
+
+   get_unapplied_transactions_result get_unapplied_transactions( const get_unapplied_transactions_params& params, const fc::time_point& deadline ) const;
+
+
+   void log_failed_transaction(const transaction_id_type& trx_id, const chain::packed_transaction_ptr& packed_trx_ptr, const char* reason) const;
 
  private:
    std::shared_ptr<class producer_plugin_impl> my;
@@ -131,3 +159,6 @@ FC_REFLECT(eosio::producer_plugin::scheduled_protocol_feature_activations, (prot
 FC_REFLECT(eosio::producer_plugin::get_supported_protocol_features_params, (exclude_disabled)(exclude_unactivatable))
 FC_REFLECT(eosio::producer_plugin::get_account_ram_corrections_params, (lower_bound)(upper_bound)(limit)(reverse))
 FC_REFLECT(eosio::producer_plugin::get_account_ram_corrections_result, (rows)(more))
+FC_REFLECT(eosio::producer_plugin::get_unapplied_transactions_params, (lower_bound)(limit)(time_limit_ms))
+FC_REFLECT(eosio::producer_plugin::unapplied_trx, (trx_id)(expiration)(trx_type)(first_auth)(first_receiver)(first_action)(total_actions)(billed_cpu_time_us)(size))
+FC_REFLECT(eosio::producer_plugin::get_unapplied_transactions_result, (size)(incoming_size)(trxs)(more))

plugins/producer_plugin/producer_plugin.cpp

@@ -1438,6 +1438,78 @@ producer_plugin::get_account_ram_corrections( const get_account_ram_corrections_
    return result;
 }
 
+producer_plugin::get_unapplied_transactions_result
+producer_plugin::get_unapplied_transactions( const get_unapplied_transactions_params& p, const fc::time_point& deadline ) const {
+
+   fc::microseconds params_time_limit = p.time_limit_ms ? fc::milliseconds(*p.time_limit_ms) : fc::milliseconds(10);
+   fc::time_point params_deadline = fc::time_point::now() + params_time_limit;
+
+   auto& ua = my->_unapplied_transactions;
+
+   auto itr = ([&](){
+      if (!p.lower_bound.empty()) {
+         try {
+            auto trx_id = transaction_id_type( p.lower_bound );
+            return ua.lower_bound( trx_id );
+         } catch( ... ) {
+            return ua.end();
+         }
+      } else {
+         return ua.begin();
+      }
+   })();
+
+   auto get_trx_type = [&](trx_enum_type t, bool read_only) {
+      if( read_only ) return "read_only";
+      switch( t ) {
+         case trx_enum_type::unknown:
+            return "unknown";
+         case trx_enum_type::persisted:
+            return "persisted";
+         case trx_enum_type::forked:
+            return "forked";
+         case trx_enum_type::aborted:
+            return "aborted";
+         case trx_enum_type::incoming:
+            return "incoming";
+      }
+      return "unknown type";
+   };
+
+   get_unapplied_transactions_result result;
+   result.size = ua.size();
+   result.incoming_size = ua.incoming_size();
+
+   uint32_t remaining = p.limit ? *p.limit : std::numeric_limits<uint32_t>::max();
+   while (itr != ua.end() && remaining > 0 && params_deadline > fc::time_point::now()) {
+      FC_CHECK_DEADLINE(deadline);
+      auto& r = result.trxs.emplace_back();
+      r.trx_id = itr->id();
+      r.expiration = itr->expiration();
+      const auto& pt = itr->trx_meta->packed_trx();
+      r.trx_type = get_trx_type( itr->trx_type, itr->trx_meta->read_only );
+      r.first_auth = pt->get_transaction().first_authorizer();
+      const auto& actions = pt->get_transaction().actions;
+      if( !actions.empty() ) {
+         r.first_receiver = actions[0].account;
+         r.first_action = actions[0].name;
+      }
+      r.total_actions = pt->get_transaction().total_actions();
+      r.billed_cpu_time_us = itr->trx_meta->billed_cpu_time_us;
+      r.size = pt->get_estimated_size();
+
+      ++itr;
+      remaining--;
+   }
+
+   if (itr != ua.end()) {
+      result.more = itr->id();
+   }
+
+   return result;
+}
+
+
 std::optional<fc::time_point> producer_plugin_impl::calculate_next_block_time(const account_name& producer_name, const block_timestamp_type& current_block_time) const {
    chain::controller& chain = chain_plug->chain();
    const auto& hbs = chain.head_block_state();

tests/plugin_http_api_test.py

@@ -1161,6 +1161,26 @@ def test_ProducerApi(self) :
         ret_json = Utils.runCmdReturnJson(valid_cmd)
         self.assertIn("rows", ret_json)
 
+        # get_unapplied_transactions with empty parameter
+        default_cmd = cmd_base + "get_unapplied_transactions"
+        ret_json = Utils.runCmdReturnJson(default_cmd)
+        self.assertIn("size", ret_json)
+        self.assertIn("incoming_size", ret_json)
+        # get_unapplied_transactions with empty content parameter
+        empty_content_cmd = default_cmd + self.http_post_str + self.empty_content_str
+        ret_json = Utils.runCmdReturnJson(empty_content_cmd)
+        self.assertIn("size", ret_json)
+        self.assertIn("incoming_size", ret_json)
+        # get_unapplied_transactions with invalid parameter
+        invalid_cmd = default_cmd + self.http_post_str + self.http_post_invalid_param
+        ret_json = Utils.runCmdReturnJson(invalid_cmd)
+        self.assertEqual(ret_json["code"], 400)
+        self.assertEqual(ret_json["error"]["code"], 3200006)
+        # get_unapplied_transactions with valid parameter
+        valid_cmd = default_cmd + self.http_post_str + ("'{\"lower_bound\":\"\", \"limit\":1, \"time_limit_ms\":500}'")
+        ret_json = Utils.runCmdReturnJson(valid_cmd)
+        self.assertIn("trxs", ret_json)
+
     # test all wallet api
     def test_WalletApi(self) :
         cmd_base = self.base_wallet_cmd_str + "wallet/"