Merge pull request #1452 from AntelopeIO/GH-1418-transport-3.2

[3.2] Ignore http error on remote_endpoint() causing terminate

.github/workflows/pinned_build.yaml

@@ -0,0 +1,59 @@
+name: "Pinned Build"
+
+on:
+  workflow_dispatch:
+
+permissions:
+  packages: read
+  contents: read
+
+defaults:
+  run:
+    shell: bash
+
+jobs:
+  Build:
+    name: Build
+    strategy:
+      fail-fast: false
+      matrix:
+        platform: [ubuntu18, ubuntu20, ubuntu22]
+    runs-on: ["self-hosted", "enf-x86-beefy-long"]
+    container: ${{ matrix.platform == 'ubuntu18' && 'ubuntu:bionic' || matrix.platform == 'ubuntu20' && 'ubuntu:focal' || 'ubuntu:jammy' }}
+    steps:
+      - name: Conditionally update git repo
+        if: ${{ matrix.platform == 'ubuntu18' }}
+        run: |
+          apt-get update
+          apt-get install -y software-properties-common
+          apt-get update
+          add-apt-repository ppa:git-core/ppa
+      - name: Update and Install git
+        run: |
+          apt-get update
+          apt-get install -y git
+          git --version
+      - name: Clone leap
+        uses: actions/checkout@v3
+        with:
+          submodules: recursive
+      - name: Install dependencies
+        run: |
+          # https://github.com/actions/runner/issues/2033
+          chown -R $(id -u):$(id -g) $PWD
+          ./scripts/install_deps.sh
+      - name: Build Pinned Build
+        env:
+          LEAP_PINNED_INSTALL_PREFIX: /usr
+        run: |
+          ./scripts/pinned_build.sh deps build "$(nproc)"
+      - name: Upload package
+        uses: actions/upload-artifact@v3
+        with:
+          name: leap-${{matrix.platform}}-pinned-amd64
+          path: build/leap_*.deb
+      - name: Run Parallel Tests
+        if: ${{ matrix.platform != 'ubuntu18' }}
+        run: |
+          cd build
+          ctest --output-on-failure -j $(nproc) -LE "(nonparallelizable_tests|long_running_tests)" --timeout 420

CMakeLists.txt

@@ -15,7 +15,7 @@ set( CXX_STANDARD_REQUIRED ON)
 
 set(VERSION_MAJOR 3)
 set(VERSION_MINOR 2)
-set(VERSION_PATCH 2)
+set(VERSION_PATCH 4)
 #set(VERSION_SUFFIX rc2)
 
 if(VERSION_SUFFIX)

libraries/chain/abi_serializer.cpp

@@ -535,14 +535,15 @@ namespace eosio { namespace chain {
             bool disallow_additional_fields = false;
             for( uint32_t i = 0; i < st.fields.size(); ++i ) {
                const auto& field = st.fields[i];
-               if( vo.contains( string(field.name).c_str() ) ) {
+               bool present = vo.contains(string(field.name).c_str());
+               if( present || is_optional(field.type) ) {
                   if( disallow_additional_fields )
                      EOS_THROW( pack_exception, "Unexpected field '${f}' found in input object while processing struct '${p}'",
                                 ("f", ctx.maybe_shorten(field.name))("p", ctx.get_path_string()) );
                   {
                      auto h1 = ctx.push_to_path( impl::field_path_item{ .parent_struct_itr = s_itr, .field_ordinal = i } );
                      auto h2 = ctx.disallow_extensions_unless( &field == &st.fields.back() );
-                     _variant_to_binary(_remove_bin_extension(field.type), vo[field.name], ds, ctx);
+                     _variant_to_binary(_remove_bin_extension(field.type), present ? vo[field.name] : fc::variant(nullptr), ds, ctx);
                   }
                } else if( ends_with(field.type, "$") && ctx.extensions_allowed() ) {
                   disallow_additional_fields = true;

libraries/chain/authorization_manager.cpp

@@ -557,8 +557,8 @@ namespace eosio { namespace chain {
 
       }
 
-      if( !allow_unused_keys  || check_but_dont_fail) {
-         EOS_ASSERT( checker.all_keys_used(), tx_irrelevant_sig,
+      if( !allow_unused_keys ) {
+         EOS_ASSERT( checker.all_keys_used() || check_but_dont_fail, tx_irrelevant_sig,
                      "transaction bears irrelevant signatures from these keys: ${keys}",
                      ("keys", checker.unused_keys()) );
       }

libraries/chain/controller.cpp

@@ -477,12 +477,13 @@ struct controller_impl {
    }
 
    void replay(std::function<bool()> check_shutdown) {
-      if( !blog.head() && !fork_db.root() ) {
+      auto blog_head = blog.head();
+      if( !fork_db.root() ) {
          fork_db.reset( *head );
-         return;
+         if (!blog_head)
+            return;
       }
 
-      auto blog_head = blog.head();
       replaying = true;
       auto start_block_num = head->block_num + 1;
       auto start = fc::time_point::now();
@@ -1153,7 +1154,7 @@ struct controller_impl {
 
       transaction_checktime_timer trx_timer(timer);
       const packed_transaction trx( std::move( etrx ) );
-      transaction_context trx_context( self, trx, std::move(trx_timer), start );
+      transaction_context trx_context( self, trx, trx.id(), std::move(trx_timer), start );
 
       trx_context.block_deadline = block_deadline;
       trx_context.max_transaction_time_subjective = max_transaction_time;
@@ -1314,7 +1315,7 @@ struct controller_impl {
       uint32_t cpu_time_to_bill_us = billed_cpu_time_us;
 
       transaction_checktime_timer trx_timer(timer);
-      transaction_context trx_context( self, *trx->packed_trx(), std::move(trx_timer) );
+      transaction_context trx_context( self, *trx->packed_trx(), gtrx.trx_id, std::move(trx_timer) );
       trx_context.leeway =  fc::microseconds(0); // avoid stealing cpu resource
       trx_context.block_deadline = block_deadline;
       trx_context.max_transaction_time_subjective = max_transaction_time;
@@ -1527,7 +1528,7 @@ struct controller_impl {
 
          const signed_transaction& trn = trx->packed_trx()->get_signed_transaction();
          transaction_checktime_timer trx_timer(timer);
-         transaction_context trx_context(self, *trx->packed_trx(), std::move(trx_timer), start, trx->read_only);
+         transaction_context trx_context(self, *trx->packed_trx(), trx->id(), std::move(trx_timer), start, trx->read_only);
          if ((bool)subjective_cpu_leeway && pending->_block_status == controller::block_status::incomplete) {
             trx_context.leeway = *subjective_cpu_leeway;
          }

libraries/chain/include/eosio/chain/transaction_context.hpp

@@ -37,6 +37,7 @@ namespace eosio { namespace chain {
 
          transaction_context( controller& c,
                               const packed_transaction& t,
+                              const transaction_id_type& trx_id, // trx_id diff than t.id() before replace_deferred
                               transaction_checktime_timer&& timer,
                               fc::time_point start = fc::time_point::now(),
                               bool read_only=false);
@@ -121,6 +122,7 @@ namespace eosio { namespace chain {
 
          controller&                                 control;
          const packed_transaction&                   packed_trx;
+         const transaction_id_type&                  id;
          std::optional<chainbase::database::session> undo_session;
          transaction_trace_ptr                       trace;
          fc::time_point                              start;

libraries/chain/transaction_context.cpp

@@ -46,11 +46,13 @@ namespace eosio { namespace chain {
 
    transaction_context::transaction_context( controller& c,
                                              const packed_transaction& t,
+                                             const transaction_id_type& trx_id,
                                              transaction_checktime_timer&& tmr,
                                              fc::time_point s,
                                              bool read_only)
    :control(c)
    ,packed_trx(t)
+   ,id(trx_id)
    ,undo_session()
    ,trace(std::make_shared<transaction_trace>())
    ,start(s)
@@ -62,7 +64,7 @@ namespace eosio { namespace chain {
       if (!c.skip_db_sessions()) {
          undo_session.emplace(c.mutable_db().start_undo_session(true));
       }
-      trace->id = packed_trx.id();
+      trace->id = id;
       trace->block_num = c.head_block_num() + 1;
       trace->block_time = c.pending_block_time();
       trace->producer_block_id = c.pending_producer_block_id();
@@ -271,7 +273,7 @@ namespace eosio { namespace chain {
          validate_referenced_accounts( trx, enforce_whiteblacklist && control.is_producing_block() );
       }
       init( initial_net_usage);
-      record_transaction( packed_trx.id(), trx.expiration ); /// checks for dupes
+      record_transaction( id, trx.expiration ); /// checks for dupes
    }
 
    void transaction_context::init_for_deferred_trx( fc::time_point p )
@@ -700,7 +702,7 @@ namespace eosio { namespace chain {
 
       uint32_t trx_size = 0;
       const auto& cgto = control.mutable_db().create<generated_transaction_object>( [&]( auto& gto ) {
-        gto.trx_id      = packed_trx.id();
+        gto.trx_id      = id;
         gto.payer       = first_auth;
         gto.sender      = account_name(); /// delayed transactions have no sender
         gto.sender_id   = transaction_id_to_sender_id( gto.trx_id );

libraries/chain/webassembly/runtimes/eos-vm.cpp

@@ -247,7 +247,7 @@ std::unique_ptr<wasm_instantiated_module_interface> eos_vm_runtime<Impl>::instan
       wasm_code_ptr code((uint8_t*)code_bytes, code_size);
       apply_options options = { .max_pages = 65536,
                                 .max_call_depth = 0 };
-      std::unique_ptr<backend_t> bkend = std::make_unique<backend_t>(code, code_size, nullptr, options);
+      std::unique_ptr<backend_t> bkend = std::make_unique<backend_t>(code, code_size, nullptr, options, false); // uses 2-passes parsing
       eos_vm_host_functions_t::resolve(bkend->get_module());
       return std::make_unique<eos_vm_instantiated_module<Impl>>(this, std::move(bkend));
    } catch(eosio::vm::exception& e) {
@@ -273,7 +273,7 @@ std::unique_ptr<wasm_instantiated_module_interface> eos_vm_profile_runtime::inst
       wasm_code_ptr code((uint8_t*)code_bytes, code_size);
       apply_options options = { .max_pages = 65536,
                                 .max_call_depth = 0 };
-      std::unique_ptr<backend_t> bkend = std::make_unique<backend_t>(code, code_size, nullptr, options);
+      std::unique_ptr<backend_t> bkend = std::make_unique<backend_t>(code, code_size, nullptr, options, false); // uses 2-passes parsing
       eos_vm_host_functions_t::resolve(bkend->get_module());
       return std::make_unique<eos_vm_profiling_module>(std::move(bkend), code_bytes, code_size);
    } catch(eosio::vm::exception& e) {

libraries/state_history/include/eosio/state_history/log.hpp

@@ -263,6 +263,9 @@ class state_history_log {
          const uint32_t num_blocks_in_log = _end_block - _begin_block;
          fc::raw::pack(log, num_blocks_in_log);
       }
+
+      log.flush();
+      index.flush();
    }
 
    // returns cfile positioned at payload

plugins/chain_plugin/include/eosio/chain_plugin/chain_plugin.hpp

@@ -821,7 +821,7 @@ public:
             // the following will convert the input to array of 2 uint128_t in little endian, i.e. 50f0fa8360ec998f4bb65b00c86282f5 fb54b91bfed2fe7fe39a92d999d002c5
             // which is the format used by secondary index
             chain::key256_t k;
-            uint8_t buffer[32];
+            uint8_t buffer[32] = {};
             boost::multiprecision::export_bits(v, buffer, 8, false);
             memcpy(&k[0], buffer + 16, 16);
             memcpy(&k[1], buffer, 16);

plugins/http_plugin/http_plugin.cpp

@@ -489,22 +489,30 @@ class http_plugin_impl : public std::enable_shared_from_this<http_plugin_impl> {
       my->plugin_state->url_handlers[url] = my->make_http_thread_url_handler(handler);
    }
 
-   void http_plugin::handle_exception( const char *api_name, const char *call_name, const string& body, const url_response_callback& cb) {
+   void http_plugin::handle_exception( const char* api_name, const char* call_name, const string& body, const url_response_callback& cb) {
       try {
          try {
             throw;
          } catch (chain::unknown_block_exception& e) {
             error_results results{400, "Unknown Block", error_results::error_info(e, verbose_http_errors)};
             cb( 400, fc::time_point::maximum(), fc::variant( results ));
+            fc_dlog( logger(), "Unknown block while processing ${api}.${call}: ${e}",
+                     ("api", api_name)("call", call_name)("e", e.to_detail_string()) );
          } catch (chain::invalid_http_request& e) {
             error_results results{400, "Invalid Request", error_results::error_info(e, verbose_http_errors)};
             cb( 400, fc::time_point::maximum(), fc::variant( results ));
+            fc_dlog( logger(), "Invalid http request while processing ${api}.${call}: ${e}",
+                     ("api", api_name)("call", call_name)("e", e.to_detail_string()) );
          } catch (chain::unsatisfied_authorization& e) {
             error_results results{401, "UnAuthorized", error_results::error_info(e, verbose_http_errors)};
             cb( 401, fc::time_point::maximum(), fc::variant( results ));
+            fc_dlog( logger(), "Auth error while processing ${api}.${call}: ${e}",
+                     ("api", api_name)("call", call_name)("e", e.to_detail_string()) );
          } catch (chain::tx_duplicate& e) {
             error_results results{409, "Conflict", error_results::error_info(e, verbose_http_errors)};
             cb( 409, fc::time_point::maximum(), fc::variant( results ));
+            fc_dlog( logger(), "Duplicate trx while processing ${api}.${call}: ${e}",
+                     ("api", api_name)("call", call_name)("e", e.to_detail_string()) );
          } catch (fc::eof_exception& e) {
             error_results results{422, "Unprocessable Entity", error_results::error_info(e, verbose_http_errors)};
             cb( 422, fc::time_point::maximum(), fc::variant( results ));

plugins/http_plugin/include/eosio/http_plugin/beast_http_listener.hpp

@@ -110,9 +110,13 @@ private:
             fail(ec, "accept", self->plugin_state_->logger, "closing connection");
          } else {
             // Create the session object and run it
+            boost::system::error_code re_ec;
+            auto re = self->socket_.remote_endpoint(re_ec);
+            std::string remote_endpoint = re_ec ? "unknown" : boost::lexical_cast<std::string>(re);
             std::make_shared<session_type>(
                   std::move(self->socket_),
-                  self->plugin_state_)
+                  self->plugin_state_,
+                  std::move(remote_endpoint))
                   ->run_session();
          }
 

plugins/http_plugin/include/eosio/http_plugin/beast_http_session.hpp

@@ -3,6 +3,9 @@
 #include <eosio/http_plugin/common.hpp>
 #include <fc/io/json.hpp>
 
+#include <boost/iostreams/device/array.hpp>
+#include <boost/iostreams/stream.hpp>
+
 #include <memory>
 #include <string>
 
@@ -75,6 +78,26 @@ bool allow_host(const http::request<http::string_body>& req, T& session,
    return true;
 }
 
+// T can be request or response or anything serializable to boost iostreams
+template<typename T>
+std::string to_log_string(const T& req, size_t max_size = 1024) {
+   assert( max_size > 4 );
+   std::string buffer( max_size, '\0' );
+   {
+      boost::iostreams::array_sink sink( buffer.data(), buffer.size() );
+      boost::iostreams::stream stream( sink );
+      stream << req;
+   }
+   buffer.resize( std::strlen( buffer.data() ) );
+   if( buffer.size() == max_size ) {
+      buffer[max_size - 3] = '.';
+      buffer[max_size - 2] = '.';
+      buffer[max_size - 1] = '.';
+   }
+   std::replace_if( buffer.begin(), buffer.end(), []( unsigned char c ) { return c == '\r' || c == '\n'; }, ' ' );
+   return buffer;
+}
+
 // Handle HTTP conneciton using boost::beast for TCP communication
 // Subclasses of this class (plain_session, ssl_session, etc.)
 // use the Curiously Recurring Template Pattern so that
@@ -95,6 +118,7 @@ protected:
    std::optional<http::response<http::string_body>> res_;
 
    std::shared_ptr<http_plugin_state> plugin_state_;
+   std::string remote_endpoint_;
 
    // whether response should be sent back to client when an exception occurs
    bool is_send_exception_response_ = true;
@@ -118,8 +142,12 @@ protected:
       }
 
       try {
-         if(!derived().allow_host(req))
+         if(!derived().allow_host(req)) {
+            error_results results{static_cast<uint16_t>(http::status::bad_request), "Disallowed HTTP HOST header in the request"};
+            send_response( fc::json::to_string( results, fc::time_point::maximum() ),
+                        static_cast<unsigned int>(http::status::bad_request) );
             return;
+         }
 
          if(!plugin_state_->access_control_allow_origin.empty()) {
             res_->set("Access-Control-Allow-Origin", plugin_state_->access_control_allow_origin);
@@ -143,6 +171,9 @@ protected:
          // verfiy bytes in flight/requests in flight
          if(!verify_max_bytes_in_flight()) return;
 
+         fc_dlog( plugin_state_->logger, "Request:  ${ep} ${r}",
+                  ("ep", remote_endpoint_)("r", to_log_string(req)) );
+
          std::string resource = std::string(req.target());
          // look for the URL handler to handle this resource
          auto handler_itr = plugin_state_->url_handlers.find(resource);
@@ -211,9 +242,9 @@ public:
    // shared_from_this() requires default constructor
    beast_http_session() = default;
 
-   beast_http_session(
-         std::shared_ptr<http_plugin_state> plugin_state)
-       : plugin_state_(std::move(plugin_state)) {
+   beast_http_session(std::shared_ptr<http_plugin_state> plugin_state, std::string remote_endpoint)
+       : plugin_state_(std::move(plugin_state)),
+         remote_endpoint_(std::move(remote_endpoint)) {
       plugin_state_->requests_in_flight += 1;
       req_parser_.emplace();
       req_parser_->body_limit(plugin_state_->max_body_size);
@@ -368,6 +399,9 @@ public:
       // Determine if we should close the connection after
       bool close = !(plugin_state_->keep_alive) || res_->need_eof();
 
+      fc_dlog( plugin_state_->logger, "Response: ${ep} ${b}",
+               ("ep", remote_endpoint_)("b", to_log_string(*res_)) );
+
       // Write the response
       auto self = derived().shared_from_this();
       http::async_write(
@@ -396,8 +430,9 @@ public:
    // Create the session
    plain_session(
          tcp_socket_t socket,
-         std::shared_ptr<http_plugin_state> plugin_state)
-       : beast_http_session<plain_session>(std::move(plugin_state)), socket_(std::move(socket)) {}
+         std::shared_ptr<http_plugin_state> plugin_state,
+         std::string remote_endpoint)
+       : beast_http_session<plain_session>(std::move(plugin_state), std::move(remote_endpoint)), socket_(std::move(socket)) {}
 
    tcp_socket_t& stream() { return socket_; }
    tcp_socket_t& socket() { return socket_; }
@@ -441,8 +476,10 @@ public:
 
    ssl_session(
          tcp_socket_t socket,
-         std::shared_ptr<http_plugin_state> plugin_state)
-       : beast_http_session<ssl_session>(std::move(plugin_state)), stream_(std::move(socket), *plugin_state_->ctx) {}
+         std::shared_ptr<http_plugin_state> plugin_state,
+         std::string remote_endpoint)
+       : beast_http_session<ssl_session>(std::move(plugin_state), std::move(remote_endpoint)),
+             stream_(std::move(socket), *plugin_state_->ctx) {}
 
 
    ssl::stream<tcp_socket_t>& stream() { return stream_; }
@@ -508,8 +545,9 @@ class unix_socket_session
 
 public:
    unix_socket_session(stream_protocol::socket sock,
-                       std::shared_ptr<http_plugin_state> plugin_state)
-       : beast_http_session(std::move(plugin_state)), socket_(std::move(sock)) {}
+                       std::shared_ptr<http_plugin_state> plugin_state,
+                       std::string remote_endpoint)
+       : beast_http_session(std::move(plugin_state), std::move(remote_endpoint)), socket_(std::move(sock)) {}
 
    virtual ~unix_socket_session() = default;
 

plugins/net_plugin/net_plugin.cpp

@@ -3166,13 +3166,13 @@ namespace eosio {
          return;
       }
 
-      bool signal_producer = !!bsp; // ready to process immediately, so signal producer to interrupt start_block
+      uint32_t block_num = bsp ? bsp->block_num : 0;
       app().post(priority::medium, [ptr{std::move(ptr)}, bsp{std::move(bsp)}, id, c = shared_from_this()]() mutable {
          c->process_signed_block( id, std::move(ptr), std::move(bsp) );
       });
 
-      if( signal_producer )
-         my_impl->producer_plug->received_block();
+      if( block_num != 0 ) // ready to process immediately, so signal producer to interrupt start_block
+         my_impl->producer_plug->received_block(block_num);
    }
 
    // called from application thread

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

@@ -145,7 +145,7 @@ public:
    void log_failed_transaction(const transaction_id_type& trx_id, const chain::packed_transaction_ptr& packed_trx_ptr, const char* reason) const;
 
    // thread-safe, called when a new block is received
-   void received_block();
+   void received_block(uint32_t block_num);
 
  private:
    std::shared_ptr<class producer_plugin_impl> my;

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

@@ -177,7 +177,10 @@ public:
       const auto time_ordinal = time_ordinal_for(now);
       const auto orig_count = _account_subjective_bill_cache.size();
       remove_subjective_billing( bsp, time_ordinal );
-      fc_dlog( log, "Subjective billed accounts ${n} removed ${r}", ("n", orig_count)("r", orig_count - _account_subjective_bill_cache.size()) );
+      if (orig_count > 0) {
+         fc_dlog( log, "Subjective billed accounts ${n} removed ${r}",
+                  ("n", orig_count)("r", orig_count - _account_subjective_bill_cache.size()) );
+      }
    }
 
    template <typename Yield>

plugins/producer_plugin/producer_plugin.cpp

@@ -333,7 +333,7 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
       std::optional<named_thread_pool>                          _thread_pool;
 
       std::atomic<int32_t>                                      _max_transaction_time_ms; // modified by app thread, read by net_plugin thread pool
-      std::atomic<bool>                                         _received_block{false}; // modified by net_plugin thread pool and app thread
+      std::atomic<uint32_t>                                     _received_block{false}; // modified by net_plugin thread pool
       fc::microseconds                                          _max_irreversible_block_age_us;
       int32_t                                                   _produce_time_offset_us = 0;
       int32_t                                                   _last_block_time_offset_us = 0;
@@ -406,8 +406,10 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
          auto before = _unapplied_transactions.size();
          _unapplied_transactions.clear_applied( bsp );
          _subjective_billing.on_block( _log, bsp, fc::time_point::now() );
-         fc_dlog( _log, "Removed applied transactions before: ${before}, after: ${after}",
-                  ("before", before)("after", _unapplied_transactions.size()) );
+         if (before > 0) {
+            fc_dlog( _log, "Removed applied transactions before: ${before}, after: ${after}",
+                     ("before", before)("after", _unapplied_transactions.size()) );
+         }
       }
 
       void on_block_header( const block_state_ptr& bsp ) {
@@ -447,7 +449,7 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
 
       bool on_incoming_block(const signed_block_ptr& block, const std::optional<block_id_type>& block_id, const block_state_ptr& bsp) {
          auto& chain = chain_plug->chain();
-         if ( _pending_block_mode == pending_block_mode::producing ) {
+         if ( in_producing_mode() ) {
             fc_wlog( _log, "dropped incoming block #${num} id: ${id}",
                      ("num", block->block_num())("id", block_id ? (*block_id).str() : "UNKNOWN") );
             return false;
@@ -461,9 +463,11 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
          const auto& id = block_id ? *block_id : block->calculate_id();
          auto blk_num = block->block_num();
 
-         fc_dlog(_log, "received incoming block ${n} ${id}", ("n", blk_num)("id", id));
+         auto now = fc::time_point::now();
+         if (now - block->timestamp < fc::minutes(5) || (blk_num % 1000 == 0)) // only log every 1000 during sync
+            fc_dlog(_log, "received incoming block ${n} ${id}", ("n", blk_num)("id", id));
 
-         EOS_ASSERT( block->timestamp < (fc::time_point::now() + fc::seconds( 7 )), block_from_the_future,
+         EOS_ASSERT( block->timestamp < (now + fc::seconds( 7 )), block_from_the_future,
                      "received a block from the future, ignoring it: ${id}", ("id", id) );
 
          /* de-dupe here... no point in aborting block if we already know the block */
@@ -512,7 +516,7 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
          }
 
          const auto& hbs = chain.head_block_state();
-         auto now = fc::time_point::now();
+         now = fc::time_point::now();
          if( hbs->header.timestamp.next().to_time_point() >= now ) {
             _production_enabled = true;
          }
@@ -595,7 +599,7 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
                   try {
                      auto result = future.get();
                      if( !self->process_incoming_transaction_async( result, persist_until_expired, return_failure_traces, next) ) {
-                        if( self->_pending_block_mode == pending_block_mode::producing ) {
+                        if( self->in_producing_mode() ) {
                            self->schedule_maybe_produce_block( true );
                         } else {
                            self->restart_speculative_block();
@@ -644,13 +648,18 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
             const auto block_deadline = calculate_block_deadline( chain.pending_block_time() );
             push_result pr = push_transaction( block_deadline, trx, persist_until_expired, return_failure_trace, next );
 
-            exhausted = pr.block_exhausted;
             if( pr.trx_exhausted ) {
                _unapplied_transactions.add_incoming( trx, persist_until_expired, return_failure_trace, next );
             } else if( pr.persist ) {
                _unapplied_transactions.add_persisted( trx );
             }
 
+            exhausted = pr.block_exhausted;
+
+            if ( !in_producing_mode() && pr.trx_exhausted )
+               exhausted = true;  // report transaction exhausted if trx was exhausted in non-producing mode (so we will restart
+                                  // a speculative block to retry it immediately, instead of waiting to receive a new block)
+
          } catch ( const guard_exception& e ) {
             chain_plugin::handle_guard_exception(e);
          } catch ( boost::interprocess::bad_alloc& ) {
@@ -693,7 +702,7 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
          exhausted
       };
 
-      inline bool should_interrupt_start_block( const fc::time_point& deadline ) const;
+      inline bool should_interrupt_start_block( const fc::time_point& deadline, uint32_t pending_block_num ) const;
       start_block_result start_block();
 
       fc::time_point calculate_pending_block_time() const;
@@ -701,6 +710,8 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
       void schedule_delayed_production_loop(const std::weak_ptr<producer_plugin_impl>& weak_this, std::optional<fc::time_point> wake_up_time);
       std::optional<fc::time_point> calculate_producer_wake_up_time( const block_timestamp_type& ref_block_time ) const;
 
+      bool in_producing_mode()   const { return _pending_block_mode == pending_block_mode::producing; }
+      bool in_speculating_mode() const { return _pending_block_mode == pending_block_mode::speculating; }
 };
 
 void new_chain_banner(const eosio::chain::controller& db)
@@ -1117,7 +1128,7 @@ void producer_plugin::resume() {
    // it is possible that we are only speculating because of this policy which we have now changed
    // re-evaluate that now
    //
-   if (my->_pending_block_mode == pending_block_mode::speculating) {
+   if (my->in_speculating_mode()) {
       my->abort_block();
       fc_ilog(_log, "Producer resumed. Scheduling production.");
       my->schedule_production_loop();
@@ -1159,7 +1170,7 @@ void producer_plugin::update_runtime_options(const runtime_options& options) {
       my->_incoming_defer_ratio = *options.incoming_defer_ratio;
    }
 
-   if (check_speculating && my->_pending_block_mode == pending_block_mode::speculating) {
+   if (check_speculating && my->in_speculating_mode()) {
       my->abort_block();
       my->schedule_production_loop();
    }
@@ -1599,7 +1610,7 @@ fc::time_point producer_plugin_impl::calculate_pending_block_time() const {
 }
 
 fc::time_point producer_plugin_impl::calculate_block_deadline( const fc::time_point& block_time ) const {
-   if( _pending_block_mode == pending_block_mode::producing ) {
+   if( in_producing_mode() ) {
       bool last_block = ((block_timestamp_type( block_time ).slot % config::producer_repetitions) == config::producer_repetitions - 1);
       return block_time + fc::microseconds(last_block ? _last_block_time_offset_us : _produce_time_offset_us);
    } else {
@@ -1607,12 +1618,12 @@ fc::time_point producer_plugin_impl::calculate_block_deadline( const fc::time_po
    }
 }
 
-bool producer_plugin_impl::should_interrupt_start_block( const fc::time_point& deadline ) const {
-   if( _pending_block_mode == pending_block_mode::producing ) {
+bool producer_plugin_impl::should_interrupt_start_block( const fc::time_point& deadline, uint32_t pending_block_num ) const {
+   if( in_producing_mode() ) {
       return deadline <= fc::time_point::now();
    }
    // if we can produce then honor deadline so production starts on time
-   return (!_producers.empty() && deadline <= fc::time_point::now()) || _received_block;
+   return (!_producers.empty() && deadline <= fc::time_point::now()) || (_received_block >= pending_block_num);
 }
 
 producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
@@ -1631,6 +1642,7 @@ producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
 
    const fc::time_point now = fc::time_point::now();
    const fc::time_point block_time = calculate_pending_block_time();
+   const uint32_t pending_block_num = hbs->block_num + 1;
    const fc::time_point preprocess_deadline = calculate_block_deadline(block_time);
 
    const pending_block_mode previous_pending_mode = _pending_block_mode;
@@ -1667,7 +1679,7 @@ producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
       _pending_block_mode = pending_block_mode::speculating;
    }
 
-   if (_pending_block_mode == pending_block_mode::producing) {
+   if (in_producing_mode()) {
       // determine if our watermark excludes us from producing at this point
       if (current_watermark) {
          const block_timestamp_type block_timestamp{block_time};
@@ -1687,16 +1699,16 @@ producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
       }
    }
 
-   if (_pending_block_mode == pending_block_mode::speculating) {
+   if (in_speculating_mode()) {
       auto head_block_age = now - chain.head_block_time();
       if (head_block_age > fc::seconds(5))
          return start_block_result::waiting_for_block;
    }
 
-   if (_pending_block_mode == pending_block_mode::producing) {
+   if (in_producing_mode()) {
       const auto start_block_time = block_time - fc::microseconds( config::block_interval_us );
       if( now < start_block_time ) {
-         fc_dlog(_log, "Not producing block waiting for production window ${n} ${bt}", ("n", hbs->block_num + 1)("bt", block_time) );
+         fc_dlog(_log, "Not producing block waiting for production window ${n} ${bt}", ("n", pending_block_num)("bt", block_time) );
          // start_block_time instead of block_time because schedule_delayed_production_loop calculates next block time from given time
          schedule_delayed_production_loop(weak_from_this(), calculate_producer_wake_up_time(start_block_time));
          return start_block_result::waiting_for_production;
@@ -1710,12 +1722,12 @@ producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
    }
 
    fc_dlog(_log, "Starting block #${n} at ${time} producer ${p}",
-           ("n", hbs->block_num + 1)("time", now)("p", scheduled_producer.producer_name));
+           ("n", pending_block_num)("time", now)("p", scheduled_producer.producer_name));
 
    try {
       uint16_t blocks_to_confirm = 0;
 
-      if (_pending_block_mode == pending_block_mode::producing) {
+      if (in_producing_mode()) {
          // determine how many blocks this producer can confirm
          // 1) if it is not a producer from this node, assume no confirmations (we will discard this block anyway)
          // 2) if it is a producer on this node that has never produced, the conservative approach is to assume no
@@ -1736,7 +1748,7 @@ producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
       abort_block();
 
       auto features_to_activate = chain.get_preactivated_protocol_features();
-      if( _pending_block_mode == pending_block_mode::producing && _protocol_features_to_activate.size() > 0 ) {
+      if( in_producing_mode() && _protocol_features_to_activate.size() > 0 ) {
          bool drop_features_to_activate = false;
          try {
             chain.validate_protocol_features( _protocol_features_to_activate );
@@ -1774,7 +1786,7 @@ producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
             std::swap( features_to_activate, protocol_features_to_activate );
             _protocol_features_signaled = true;
             ilog( "signaling activation of the following protocol features in block ${num}: ${features_to_activate}",
-                  ("num", hbs->block_num + 1)("features_to_activate", features_to_activate) );
+                  ("num", pending_block_num)("features_to_activate", features_to_activate) );
          }
       }
 
@@ -1784,7 +1796,7 @@ producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
    if( chain.is_building_block() ) {
       const auto& pending_block_signing_authority = chain.pending_block_signing_authority();
 
-      if (_pending_block_mode == pending_block_mode::producing && pending_block_signing_authority != scheduled_producer.authority) {
+      if (in_producing_mode() && pending_block_signing_authority != scheduled_producer.authority) {
          elog("Unexpected block signing authority, reverting to speculative mode! [expected: \"${expected}\", actual: \"${actual\"", ("expected", scheduled_producer.authority)("actual", pending_block_signing_authority));
          _pending_block_mode = pending_block_mode::speculating;
       }
@@ -1798,7 +1810,7 @@ producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
          if( !remove_expired_blacklisted_trxs( preprocess_deadline ) )
             return start_block_result::exhausted;
          if( !_subjective_billing.remove_expired( _log, chain.pending_block_time(), fc::time_point::now(),
-                                                  [&](){ return should_interrupt_start_block( preprocess_deadline ); } ) ) {
+                                                  [&](){ return should_interrupt_start_block( preprocess_deadline, pending_block_num ); } ) ) {
             return start_block_result::exhausted;
          }
 
@@ -1808,7 +1820,7 @@ producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
          if( !process_unapplied_trxs( preprocess_deadline ) )
             return start_block_result::exhausted;
 
-         if (_pending_block_mode == pending_block_mode::producing) {
+         if (in_producing_mode()) {
             auto scheduled_trx_deadline = preprocess_deadline;
             if (_max_scheduled_transaction_time_per_block_ms >= 0) {
                scheduled_trx_deadline = std::min<fc::time_point>(
@@ -1822,7 +1834,7 @@ producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
 
          if( app().is_quiting() ) // db guard exception above in LOG_AND_DROP could have called app().quit()
             return start_block_result::failed;
-         if ( should_interrupt_start_block( preprocess_deadline ) || block_is_exhausted() ) {
+         if ( should_interrupt_start_block( preprocess_deadline, pending_block_num ) || block_is_exhausted() ) {
             return start_block_result::exhausted;
          }
 
@@ -1849,12 +1861,13 @@ bool producer_plugin_impl::remove_expired_trxs( const fc::time_point& deadline )
 {
    chain::controller& chain = chain_plug->chain();
    auto pending_block_time = chain.pending_block_time();
+   auto pending_block_num = chain.pending_block_num();
 
    // remove all expired transactions
    size_t num_expired_persistent = 0;
    size_t num_expired_other = 0;
    size_t orig_count = _unapplied_transactions.size();
-   bool exhausted = !_unapplied_transactions.clear_expired( pending_block_time, [&](){ return should_interrupt_start_block(deadline); },
+   bool exhausted = !_unapplied_transactions.clear_expired( pending_block_time, [&](){ return should_interrupt_start_block(deadline, pending_block_num); },
          [&num_expired_persistent, &num_expired_other]( const packed_transaction_ptr& packed_trx_ptr, trx_enum_type trx_type ) {
             // expired exception is logged as part of next() call
             if( trx_type == trx_enum_type::persisted ) {
@@ -1864,7 +1877,7 @@ bool producer_plugin_impl::remove_expired_trxs( const fc::time_point& deadline )
             }
    });
 
-   if( exhausted && _pending_block_mode == pending_block_mode::producing ) {
+   if( exhausted && in_producing_mode() ) {
       fc_wlog( _log, "Unable to process all expired transactions in unapplied queue before deadline, "
                      "Persistent expired ${persistent_expired}, Other expired ${other_expired}",
                ("persistent_expired", num_expired_persistent)("other_expired", num_expired_other) );
@@ -1885,12 +1898,13 @@ bool producer_plugin_impl::remove_expired_blacklisted_trxs( const fc::time_point
    if(!blacklist_by_expiry.empty()) {
       const chain::controller& chain = chain_plug->chain();
       const auto lib_time = chain.last_irreversible_block_time();
+      const auto pending_block_num = chain.pending_block_num();
 
       int num_expired = 0;
       int orig_count = _blacklisted_transactions.size();
 
       while (!blacklist_by_expiry.empty() && blacklist_by_expiry.begin()->expiry <= lib_time) {
-         if ( should_interrupt_start_block( deadline ) ) {
+         if ( should_interrupt_start_block( deadline, pending_block_num ) ) {
             exhausted = true;
             break;
          }
@@ -1966,7 +1980,7 @@ void producer_plugin_impl::log_trx_results( const packed_transaction_ptr& trx,
 
    bool except = except_ptr || (trace && trace->except);
    if (except) {
-      if (_pending_block_mode == pending_block_mode::producing) {
+      if (in_producing_mode()) {
          fc_dlog(_trx_log, "[TRX_TRACE] Block ${block_num} for producer ${prod} is REJECTING tx: ${trx}",
                  ("block_num", chain.head_block_num() + 1)("prod", get_pending_block_producer())
                  ("trx", chain_plug->get_log_trx(trx->get_transaction())));
@@ -1987,7 +2001,7 @@ void producer_plugin_impl::log_trx_results( const packed_transaction_ptr& trx,
                  ("entire_trace", get_trace(trace, except_ptr)));
       }
    } else {
-      if (_pending_block_mode == pending_block_mode::producing) {
+      if (in_producing_mode()) {
          fc_dlog(_trx_log, "[TRX_TRACE] Block ${block_num} for producer ${prod} is ACCEPTING tx: ${trx}",
                  ("block_num", chain.head_block_num() + 1)("prod", get_pending_block_producer())
                  ("trx", chain_plug->get_log_trx(trx->get_transaction())));
@@ -2041,15 +2055,14 @@ producer_plugin_impl::push_transaction( const fc::time_point& block_deadline,
    fc::microseconds max_trx_time = fc::milliseconds( _max_transaction_time_ms.load() );
    if( max_trx_time.count() < 0 ) max_trx_time = fc::microseconds::maximum();
 
-   bool disable_subjective_billing = ( _pending_block_mode == pending_block_mode::producing )
-                                     || disable_subjective_enforcement;
+   bool disable_subjective_billing = in_producing_mode() || disable_subjective_enforcement;
 
    int64_t sub_bill = 0;
    if( !disable_subjective_billing )
       sub_bill = _subjective_billing.get_subjective_bill( first_auth, fc::time_point::now() );
 
    auto prev_billed_cpu_time_us = trx->billed_cpu_time_us;
-   if( _pending_block_mode == pending_block_mode::producing && prev_billed_cpu_time_us > 0 ) {
+   if( in_producing_mode() && prev_billed_cpu_time_us > 0 ) {
       const auto& rl = chain.get_resource_limits_manager();
       if ( !_subjective_billing.is_account_disabled( first_auth ) && !rl.is_unlimited_cpu( first_auth ) ) {
          int64_t prev_billed_plus100_us = prev_billed_cpu_time_us + EOS_PERCENT( prev_billed_cpu_time_us, 100 * config::percent_1 );
@@ -2063,7 +2076,7 @@ producer_plugin_impl::push_transaction( const fc::time_point& block_deadline,
    if( trace->except ) {
       _time_tracker.add_fail_time(end - start);
       if( exception_is_exhausted( *trace->except ) ) {
-         if( _pending_block_mode == pending_block_mode::producing ) {
+         if( in_producing_mode() ) {
             fc_dlog(_trx_failed_trace_log, "[TRX_TRACE] Block ${block_num} for producer ${prod} COULD NOT FIT, tx: ${txid} RETRYING ",
                     ("block_num", chain.head_block_num() + 1)("prod", get_pending_block_producer())("txid", trx->id()));
          } else {
@@ -2124,15 +2137,15 @@ bool producer_plugin_impl::process_unapplied_trxs( const fc::time_point& deadlin
    bool exhausted = false;
    if( !_unapplied_transactions.empty() ) {
       if( _pending_block_mode != pending_block_mode::producing && _disable_persist_until_expired ) return !exhausted;
+      const chain::controller& chain = chain_plug->chain();
+      const auto pending_block_num = chain.pending_block_num();
       int num_applied = 0, num_failed = 0, num_processed = 0;
       auto unapplied_trxs_size = _unapplied_transactions.size();
       // unapplied and persisted do not have a next method to call
-      auto itr     = (_pending_block_mode == pending_block_mode::producing) ?
-                     _unapplied_transactions.unapplied_begin() : _unapplied_transactions.persisted_begin();
-      auto end_itr = (_pending_block_mode == pending_block_mode::producing) ?
-                     _unapplied_transactions.unapplied_end()   : _unapplied_transactions.persisted_end();
+      auto itr     = in_producing_mode() ? _unapplied_transactions.unapplied_begin() : _unapplied_transactions.persisted_begin();
+      auto end_itr = in_producing_mode() ? _unapplied_transactions.unapplied_end()   : _unapplied_transactions.persisted_end();
       while( itr != end_itr ) {
-         if( should_interrupt_start_block( deadline ) ) {
+         if( should_interrupt_start_block( deadline, pending_block_num ) ) {
             exhausted = true;
             break;
          }
@@ -2313,8 +2326,10 @@ bool producer_plugin_impl::process_incoming_trxs( const fc::time_point& deadline
    if( itr != end ) {
       size_t processed = 0;
       fc_dlog( _log, "Processing ${n} pending transactions", ("n", _unapplied_transactions.incoming_size()) );
+      const chain::controller& chain = chain_plug->chain();
+      const auto pending_block_num = chain.pending_block_num();
       while( itr != end ) {
-         if ( should_interrupt_start_block( deadline ) ) {
+         if ( should_interrupt_start_block( deadline, pending_block_num ) ) {
             exhausted = true;
             break;
          }
@@ -2360,7 +2375,6 @@ bool producer_plugin_impl::block_is_exhausted() const {
 // -> Idle
 // --> Start block B (block time y.000) at time x.500
 void producer_plugin_impl::schedule_production_loop() {
-   _received_block = false;
    _timer.cancel();
 
    auto result = start_block();
@@ -2384,17 +2398,17 @@ void producer_plugin_impl::schedule_production_loop() {
          fc_dlog(_log, "Waiting till another block is received and scheduling Speculative/Production Change");
          schedule_delayed_production_loop(weak_from_this(), calculate_producer_wake_up_time(calculate_pending_block_time()));
       } else {
-         fc_dlog(_log, "Waiting till another block is received");
+         fc_tlog(_log, "Waiting till another block is received");
          // nothing to do until more blocks arrive
       }
 
    } else if (result == start_block_result::waiting_for_production) {
       // scheduled in start_block()
 
-   } else if (_pending_block_mode == pending_block_mode::producing) {
+   } else if (in_producing_mode()) {
       schedule_maybe_produce_block( result == start_block_result::exhausted );
 
-   } else if (_pending_block_mode == pending_block_mode::speculating && !_producers.empty() && !production_disabled_by_policy()){
+   } else if (in_speculating_mode() && !_producers.empty() && !production_disabled_by_policy()){
       chain::controller& chain = chain_plug->chain();
       fc_dlog(_log, "Speculative Block Created; Scheduling Speculative/Production Change");
       EOS_ASSERT( chain.is_building_block(), missing_pending_block_state, "speculating without pending_block_state" );
@@ -2511,7 +2525,7 @@ static auto maybe_make_debug_time_logger() -> std::optional<decltype(make_debug_
 void producer_plugin_impl::produce_block() {
    //ilog("produce_block ${t}", ("t", fc::time_point::now())); // for testing _produce_time_offset_us
    auto start = fc::time_point::now();
-   EOS_ASSERT(_pending_block_mode == pending_block_mode::producing, producer_exception, "called produce_block while not actually producing");
+   EOS_ASSERT(in_producing_mode(), producer_exception, "called produce_block while not actually producing");
    chain::controller& chain = chain_plug->chain();
    EOS_ASSERT(chain.is_building_block(), missing_pending_block_state, "pending_block_state does not exist but it should, another plugin may have corrupted it");
 
@@ -2565,8 +2579,8 @@ void producer_plugin_impl::produce_block() {
         ("confs", new_bs->header.confirmed));
 }
 
-void producer_plugin::received_block() {
-   my->_received_block = true;
+void producer_plugin::received_block(uint32_t block_num) {
+   my->_received_block = block_num;
 }
 
 void producer_plugin::log_failed_transaction(const transaction_id_type& trx_id, const packed_transaction_ptr& packed_trx_ptr, const char* reason) const {

plugins/state_history_plugin/state_history_plugin.cpp

@@ -351,6 +351,7 @@ struct state_history_plugin_impl : std::enable_shared_from_this<state_history_pl
 
          get_blocks_result_v0 result;
          result.head = {block_state->block_num, block_state->id};
+         to_send_block_num = std::min(block_state->block_num, to_send_block_num);
          send_update(std::move(result), block_state);
       }
 

programs/leap-util/actions/blocklog.cpp

@@ -88,7 +88,7 @@ void blocklog_actions::setup(CLI::App& app) {
    extract_blocks->add_option("--blocks-dir", opt->blocks_dir, "The location of the blocks directory (absolute path or relative to the current directory).");
    extract_blocks->add_option("--first,-f", opt->first_block, "The first block number to keep.")->required();
    extract_blocks->add_option("--last,-l", opt->last_block, "The last block number to keep.")->required();
-   extract_blocks->add_option("--output-dir", opt->output_dir, "The output directory for the block log extracted from blocks-dir.");
+   extract_blocks->add_option("--output-dir", opt->output_dir, "The output directory for the block log extracted from blocks-dir.")->required();
 
    // subcommand - smoke test
    auto* smoke_test = sub->add_subcommand("smoke-test", "Quick test that blocks.log and blocks.index are well formed and agree with each other.")->callback([err_guard]() { err_guard(&blocklog_actions::smoke_test); });

scripts/pinned_build.sh

@@ -132,7 +132,7 @@ pushdir ${LEAP_DIR}
 
 # build Leap
 echo "Building Leap ${SCRIPT_DIR}"
-try cmake -DCMAKE_TOOLCHAIN_FILE=${SCRIPT_DIR}/pinned_toolchain.cmake -DCMAKE_INSTALL_PREFIX=/usr/local -DCMAKE_BUILD_TYPE=Release -DCMAKE_PREFIX_PATH=${LLVM_DIR}/lib/cmake -DCMAKE_PREFIX_PATH=${BOOST_DIR}/bin ${SCRIPT_DIR}/..
+try cmake -DCMAKE_TOOLCHAIN_FILE=${SCRIPT_DIR}/pinned_toolchain.cmake -DCMAKE_INSTALL_PREFIX=${LEAP_PINNED_INSTALL_PREFIX:-/usr/local} -DCMAKE_BUILD_TYPE=Release -DCMAKE_PREFIX_PATH=${LLVM_DIR}/lib/cmake -DCMAKE_PREFIX_PATH=${BOOST_DIR}/bin ${SCRIPT_DIR}/..
 
 try make -j${JOBS}
 try cpack

tests/CMakeLists.txt

@@ -98,13 +98,13 @@ target_link_libraries(ship_client abieos Boost::program_options Boost::system Th
 add_executable(ship_streamer ship_streamer.cpp)
 target_link_libraries(ship_streamer abieos Boost::program_options Boost::system Threads::Threads)
 
-add_test(NAME ship_test COMMAND tests/ship_test.py -v --num-clients 1 --num-requests 5000 --clean-run --dump-error-detail WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
+add_test(NAME ship_test COMMAND tests/ship_test.py -v --num-clients 10 --num-requests 5000 --clean-run --dump-error-detail WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
 set_property(TEST ship_test PROPERTY LABELS nonparallelizable_tests)
-add_test(NAME ship_test_unix COMMAND tests/ship_test.py -v --num-clients 1 --num-requests 5000 --clean-run --dump-error-detail --unix-socket WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
+add_test(NAME ship_test_unix COMMAND tests/ship_test.py -v --num-clients 10 --num-requests 5000 --clean-run --dump-error-detail --unix-socket WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
 set_property(TEST ship_test_unix PROPERTY LABELS nonparallelizable_tests)
 
-add_test(NAME ship_streamer_test COMMAND tests/ship_streamer_test.py -v --num-clients 1 --num-blocks 50 --clean-run --dump-error-detail WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
-set_property(TEST ship_streamer_test PROPERTY LABELS nonparallelizable_tests)
+add_test(NAME ship_streamer_test COMMAND tests/ship_streamer_test.py -v --num-clients 10 --clean-run --dump-error-detail WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
+set_property(TEST ship_streamer_test PROPERTY LABELS long_running_tests)
 
 add_test(NAME p2p_dawn515_test COMMAND tests/p2p_tests/dawn_515/test.sh WORKING_DIRECTORY ${CMAKE_BINARY_DIR})
 set_property(TEST p2p_dawn515_test PROPERTY LABELS nonparallelizable_tests)

tests/http_plugin_test.py

@@ -1,6 +1,8 @@
 #!/usr/bin/env python3
 
 from TestHarness import Cluster, Node, ReturnType, TestHelper, Utils
+import urllib.request
+import sys
 
 ###############################################################
 # http_plugin_tests.py
@@ -23,7 +25,7 @@ dumpErrorDetails = dumpErrorDetails=args.dump_error_details
 
 Utils.Debug=debug
 https_port = 5555
-cluster=Cluster(walletd=True)
+cluster=Cluster(host="127.0.0.1", walletd=True)
 
 testSuccessful=False
 
@@ -35,21 +37,34 @@ try:
     TestHelper.printSystemInfo("BEGIN")
 
     Print("Stand up cluster")
-    # standup cluster with HTTPS enabled, but not configured
-    # HTTP should still work
-    extraArgs={ 0 : "--https-server-address 127.0.0.1:5555" }
-    # specificExtraNodeosArgs=extraArgs
 
-    if cluster.launch(dontBootstrap=True, loadSystemContract=False) is False:
+    if cluster.launch(dontBootstrap=True, loadSystemContract=False, specificExtraNodeosArgs = {0: "--http-validate-host true"}) is False:
         cmdError("launcher")
         errorExit("Failed to stand up eos cluster.")
-
     Print("Getting cluster info")
     cluster.getInfos()
+
+    node0 = cluster.nodes[0]
+    ## HTTP plugin listens to 127.0.0.1:8888 by default. With the --http-validate-host=true,
+    ## the HTTP request to "http://localhost:8888" should fail because the HOST header doesn't
+    ## match "127.0.0.1".
+
+    def get_info_status(url):
+        try:
+            req = urllib.request.Request( url, method = "GET")
+            return urllib.request.urlopen(req, data=None).code
+        except urllib.error.HTTPError as response:
+            return response.code
+        except:
+            e = sys.exc_info()[0]
+            return e
+    url = node0.endpointHttp.replace("127.0.0.1", "localhost") + "/v1/chain/get_info"
+    code = get_info_status(url)
+    assert code == 400, f"Expected HTTP returned code 400, got {code}"
     testSuccessful = True
+
 finally:
     TestHelper.shutdown(cluster, None, testSuccessful, killEosInstances, True, keepLogs, killAll, dumpErrorDetails)
 
 exitCode = 0 if testSuccessful else 1
 exit(exitCode)
-        

tests/ship_client.cpp

@@ -125,6 +125,8 @@ int main(int argc, char* argv[]) {
             eosio::check(result_doucment[1]["head"].IsObject(),                                 "'head' is not an object");
             eosio::check(result_doucment[1]["head"].HasMember("block_num"),                     "'head' does not contain 'block_num'");
             eosio::check(result_doucment[1]["head"]["block_num"].IsUint(),                      "'head.block_num' isn't a number");
+            eosio::check(result_doucment[1]["head"].HasMember("block_id"),                      "'head' does not contain 'block_id'");
+            eosio::check(result_doucment[1]["head"]["block_id"].IsString(),                     "'head.block_id' isn't a string");
 
             uint32_t this_block_num = result_doucment[1]["head"]["block_num"].GetUint();
 

tests/ship_streamer.cpp

@@ -110,30 +110,61 @@ int main(int argc, char* argv[]) {
       stream.binary(true);
       stream.write(boost::asio::buffer(request_type.json_to_bin(request_sb.GetString(), [](){})));
 
+      //      block_num, block_id
+      std::map<uint32_t, std::string> block_ids;
       bool is_first = true;
       for(;;) {
          boost::beast::flat_buffer buffer;
          stream.read(buffer);
 
          eosio::input_stream is((const char*)buffer.data().data(), buffer.data().size());
-         rapidjson::Document result_doucment;
-         result_doucment.Parse(result_type.bin_to_json(is).c_str());
+         rapidjson::Document result_document;
+         result_document.Parse(result_type.bin_to_json(is).c_str());
 
-         eosio::check(!result_doucment.HasParseError(),                                      "Failed to parse result JSON from abieos");
-         eosio::check(result_doucment.IsArray(),                                             "result should have been an array (variant) but it's not");
-         eosio::check(result_doucment.Size() == 2,                                           "result was an array but did not contain 2 items like a variant should");
-         eosio::check(std::string(result_doucment[0].GetString()) == "get_blocks_result_v0", "result type doesn't look like get_blocks_result_v0");
-         eosio::check(result_doucment[1].IsObject(),                                         "second item in result array is not an object");
-         eosio::check(result_doucment[1].HasMember("head"),                                  "cannot find 'head' in result");
-         eosio::check(result_doucment[1]["head"].IsObject(),                                 "'head' is not an object");
-         eosio::check(result_doucment[1]["head"].HasMember("block_num"),                     "'head' does not contain 'block_num'");
-         eosio::check(result_doucment[1]["head"]["block_num"].IsUint(),                      "'head.block_num' isn't a number");
+         eosio::check(!result_document.HasParseError(),                                      "Failed to parse result JSON from abieos");
+         eosio::check(result_document.IsArray(),                                             "result should have been an array (variant) but it's not");
+         eosio::check(result_document.Size() == 2,                                           "result was an array but did not contain 2 items like a variant should");
+         eosio::check(std::string(result_document[0].GetString()) == "get_blocks_result_v0", "result type doesn't look like get_blocks_result_v0");
+         eosio::check(result_document[1].IsObject(),                                         "second item in result array is not an object");
+         eosio::check(result_document[1].HasMember("head"),                                  "cannot find 'head' in result");
+         eosio::check(result_document[1]["head"].IsObject(),                                 "'head' is not an object");
+         eosio::check(result_document[1]["head"].HasMember("block_num"),                     "'head' does not contain 'block_num'");
+         eosio::check(result_document[1]["head"]["block_num"].IsUint(),                      "'head.block_num' isn't a number");
+         eosio::check(result_document[1]["head"].HasMember("block_id"),                      "'head' does not contain 'block_id'");
+         eosio::check(result_document[1]["head"]["block_id"].IsString(),                     "'head.block_id' isn't a string");
 
          uint32_t this_block_num = 0;
-         if( result_doucment[1].HasMember("this_block") && result_doucment[1]["this_block"].IsObject() ) {
-            if( result_doucment[1]["this_block"].HasMember("block_num") && result_doucment[1]["this_block"]["block_num"].IsUint() ) {
-               this_block_num = result_doucment[1]["this_block"]["block_num"].GetUint();
+         if( result_document[1].HasMember("this_block") && result_document[1]["this_block"].IsObject() ) {
+            if( result_document[1]["this_block"].HasMember("block_num") && result_document[1]["this_block"]["block_num"].IsUint() ) {
+               this_block_num = result_document[1]["this_block"]["block_num"].GetUint();
             }
+            std::string this_block_id;
+            if( result_document[1]["this_block"].HasMember("block_id") && result_document[1]["this_block"]["block_id"].IsString() ) {
+               this_block_id = result_document[1]["this_block"]["block_id"].GetString();
+            }
+            std::string prev_block_id;
+            if( result_document[1]["prev_block"].HasMember("block_id") && result_document[1]["prev_block"]["block_id"].IsString() ) {
+               prev_block_id = result_document[1]["prev_block"]["block_id"].GetString();
+            }
+            if( !irreversible_only && !this_block_id.empty() && !prev_block_id.empty() ) {
+               // verify forks were sent
+               if (block_ids.count(this_block_num-1)) {
+                  if (block_ids[this_block_num-1] != prev_block_id) {
+                     std::cerr << "Received block: << " << this_block_num << " that does not link to previous: " << block_ids[this_block_num-1] << std::endl;
+                     return 1;
+                  }
+               }
+               block_ids[this_block_num] = this_block_id;
+
+               if( result_document[1]["last_irreversible"].HasMember("block_num") && result_document[1]["last_irreversible"]["block_num"].IsUint() ) {
+                  uint32_t lib_num = result_document[1]["last_irreversible"]["block_num"].GetUint();
+                  auto i = block_ids.lower_bound(lib_num);
+                  if (i != block_ids.end()) {
+                     block_ids.erase(block_ids.begin(), i);
+                  }
+               }
+            }
+
          }
 
          if(is_first) {
@@ -146,7 +177,7 @@ int main(int argc, char* argv[]) {
 
          rapidjson::StringBuffer result_sb;
          rapidjson::PrettyWriter<rapidjson::StringBuffer> result_writer(result_sb);
-         result_doucment[1].Accept(result_writer);
+         result_document[1].Accept(result_writer);
          std::cout << result_sb.GetString() << std::endl << "}" << std::endl;
 
          if( this_block_num == end_block_num ) break;

tests/ship_streamer_test.py

@@ -9,32 +9,32 @@ import sys
 
 from TestHarness import Cluster, TestHelper, Utils, WalletMgr
 from TestHarness.TestHelper import AppArgs
+from core_symbol import CORE_SYMBOL
 
 ###############################################################
 # ship_streamer_test
 # 
-# This test sets up <-p> producing node(s) and <-n - -p>
-#   non-producing node(s). One of the non-producing nodes
-#   is configured with the state_history_plugin.  An instance
-#   of node will be started with ship_streamer to exercise
-#   the SHiP API.
+# This test sets up 2 producing nodes and one "bridge" node using test_control_api_plugin.
+#   One producing node has 3 of the elected producers and the other has 1 of the elected producers.
+#   All the producers are named in alphabetical order, so that the 3 producers, in the one production node, are
+#       scheduled first, followed by the 1 producer in the other producer node. Each producing node is only connected
+#       to the other producing node via the "bridge" node.
+#   The bridge node has the test_control_api_plugin, that the test uses to kill
+#       the "bridge" node to generate a fork.
+#   ship_streamer is used to connect to the state_history_plugin and verify that blocks receive link to previous
+#   blocks. If the blocks do not link then ship_streamer will exit with an error causing this test to generate an
+#   error. The fork generated by nodeos should be sent to the ship_streamer so it is able to correctly observe the
+#   fork.
 #
 ###############################################################
 
 Print=Utils.Print
 
 appArgs = AppArgs()
-extraArgs = appArgs.add(flag="--num-blocks", type=int, help="How many blocsk to stream from ship_streamer", default=20)
 extraArgs = appArgs.add(flag="--num-clients", type=int, help="How many ship_streamers should be started", default=1)
-args = TestHelper.parse_args({"-p", "-n","--dump-error-details","--keep-logs","-v","--leave-running","--clean-run"}, applicationSpecificArgs=appArgs)
+args = TestHelper.parse_args({"--dump-error-details","--keep-logs","-v","--leave-running","--clean-run"}, applicationSpecificArgs=appArgs)
 
 Utils.Debug=args.v
-totalProducerNodes=args.p
-totalNodes=args.n
-if totalNodes<=totalProducerNodes:
-    totalNodes=totalProducerNodes+1
-totalNonProducerNodes=totalNodes-totalProducerNodes
-totalProducers=totalProducerNodes
 cluster=Cluster(walletd=True)
 dumpErrorDetails=args.dump_error_details
 keepLogs=args.keep_logs
@@ -42,6 +42,12 @@ dontKill=args.leave_running
 killAll=args.clean_run
 walletPort=TestHelper.DEFAULT_WALLET_PORT
 
+totalProducerNodes=2
+totalNonProducerNodes=1
+totalNodes=totalProducerNodes+totalNonProducerNodes
+maxActiveProducers=21
+totalProducers=maxActiveProducers
+
 walletMgr=WalletMgr(True, port=walletPort)
 testSuccessful=False
 killEosInstances=not dontKill
@@ -57,12 +63,20 @@ try:
     cluster.killall(allInstances=killAll)
     cluster.cleanup()
     Print("Stand up cluster")
-    specificExtraNodeosArgs={}
-    # non-producing nodes are at the end of the cluster's nodes, so reserving the last one for state_history_plugin
-    shipNodeNum = totalNodes - 1
-    specificExtraNodeosArgs[shipNodeNum]="--plugin eosio::state_history_plugin --disable-replay-opts --trace-history --sync-fetch-span 200 --plugin eosio::net_api_plugin "
 
-    if cluster.launch(pnodes=totalProducerNodes,
+
+    # ***   setup topogrophy   ***
+
+    # "bridge" shape connects defprocera through defproducerc (3 in node0) to each other and defproduceru (1 in node1)
+    # and the only connection between those 2 groups is through the bridge node
+
+    shipNodeNum = 1
+    specificExtraNodeosArgs={}
+    specificExtraNodeosArgs[shipNodeNum]="--plugin eosio::state_history_plugin --disable-replay-opts --trace-history --chain-state-history --plugin eosio::net_api_plugin "
+    # producer nodes will be mapped to 0 through totalProducerNodes-1, so the number totalProducerNodes will be the non-producing node
+    specificExtraNodeosArgs[totalProducerNodes]="--plugin eosio::test_control_api_plugin  "
+
+    if cluster.launch(topo="bridge", pnodes=totalProducerNodes,
                       totalNodes=totalNodes, totalProducers=totalProducers,
                       useBiosBootFile=False, specificExtraNodeosArgs=specificExtraNodeosArgs) is False:
         Utils.cmdError("launcher")
@@ -70,19 +84,81 @@ try:
 
     # ***   identify each node (producers and non-producing node)   ***
 
-    shipNode = cluster.getNode(shipNodeNum)
-    prodNode = cluster.getNode(0)
-
     #verify nodes are in sync and advancing
     cluster.waitOnClusterSync(blockAdvancing=5)
     Print("Cluster in Sync")
 
+    prodNode = cluster.getNode(0)
+    prodNode0 = prodNode
+    prodNode1 = cluster.getNode(1)
+    nonProdNode = cluster.getNode(2)
+    shipNode = cluster.getNode(shipNodeNum)
+
+
+    accounts=cluster.createAccountKeys(6)
+    if accounts is None:
+        Utils.errorExit("FAILURE - create keys")
+
+    accounts[0].name="testeraaaaaa"
+    accounts[1].name="tester111111" # needed for voting
+    accounts[2].name="tester222222" # needed for voting
+    accounts[3].name="tester333333" # needed for voting
+    accounts[4].name="tester444444" # needed for voting
+    accounts[5].name="tester555555" # needed for voting
+
+    testWalletName="test"
+
+    Print(f"Creating wallet {testWalletName}.")
+    testWallet=walletMgr.create(testWalletName, [cluster.eosioAccount,accounts[0],accounts[1],accounts[2],accounts[3],accounts[4],accounts[5]])
+
+    for _, account in cluster.defProducerAccounts.items():
+        walletMgr.importKey(account, testWallet, ignoreDupKeyWarning=True)
+
+    for i in range(0, totalNodes):
+        node=cluster.getNode(i)
+        node.producers=Cluster.parseProducers(i)
+        for prod in node.producers:
+            prodName = cluster.defProducerAccounts[prod].name
+            if prodName == "defproducera" or prodName == "defproducerb" or prodName == "defproducerc" or prodName == "defproduceru":
+                Print(f"Register producer {prodName}")
+                trans=node.regproducer(cluster.defProducerAccounts[prod], "http://mysite.com", 0, waitForTransBlock=False, exitOnError=True)
+
+    # create accounts via eosio as otherwise a bid is needed
+    for account in accounts:
+        Print(f"Create new account {account.name} via {cluster.eosioAccount.name} with private key: {account.activePrivateKey}")
+        trans=nonProdNode.createInitializeAccount(account, cluster.eosioAccount, stakedDeposit=0, waitForTransBlock=True, stakeNet=10000, stakeCPU=10000, buyRAM=10000000, exitOnError=True)
+        transferAmount="100000000.0000 {0}".format(CORE_SYMBOL)
+        Print(f"Transfer funds {transferAmount} from account {cluster.eosioAccount.name} to {account.name}")
+        nonProdNode.transferFunds(cluster.eosioAccount, account, transferAmount, "test transfer", waitForTransBlock=True)
+        trans=nonProdNode.delegatebw(account, 20000000.0000, 20000000.0000, waitForTransBlock=False, exitOnError=True)
+
+    # ***   vote using accounts   ***
+
+    cluster.waitOnClusterSync(blockAdvancing=3)
     start_block_num = shipNode.getBlockNum()
-    end_block_num = start_block_num + args.num_blocks
+
+    # vote a,b,c (node0)  u (node1)
+    voteProducers=[]
+    voteProducers.append("defproducera")
+    voteProducers.append("defproducerb")
+    voteProducers.append("defproducerc")
+    voteProducers.append("defproduceru")
+    for account in accounts:
+        Print(f"Account {account.name} vote for producers={voteProducers}")
+        trans=prodNode.vote(account, voteProducers, exitOnError=True, waitForTransBlock=False)
+
+    #verify nodes are in sync and advancing
+    cluster.waitOnClusterSync(blockAdvancing=3)
+    Print("Shutdown unneeded bios node")
+    cluster.biosNode.kill(signal.SIGTERM)
+    prodNode0.waitForProducer("defproducerc")
+
+    block_range = 350
+    end_block_num = start_block_num + block_range
 
     shipClient = "tests/ship_streamer"
-    cmd = "%s --start-block-num %d --end-block-num %d --fetch-block --fetch-traces --fetch-deltas" % (shipClient, start_block_num, end_block_num)
-    if Utils.Debug: Utils.Print("cmd: %s" % (cmd))
+    cmd = f"{shipClient} --start-block-num {start_block_num} --end-block-num {end_block_num} --fetch-block --fetch-traces --fetch-deltas"
+    if Utils.Debug: Utils.Print(f"cmd: {cmd}")
     clients = []
     files = []
     shipTempDir = os.path.join(Utils.DataDir, "ship")
@@ -92,28 +168,61 @@ try:
     starts = []
     for i in range(0, args.num_clients):
         start = time.perf_counter()
-        outFile = open("%s%d.out" % (shipClientFilePrefix, i), "w")
-        errFile = open("%s%d.err" % (shipClientFilePrefix, i), "w")
-        Print("Start client %d" % (i))
+        outFile = open(f"{shipClientFilePrefix}{i}.out", "w")
+        errFile = open(f"{shipClientFilePrefix}{i}.err", "w")
+        Print(f"Start client {i}")
         popen=Utils.delayedCheckOutput(cmd, stdout=outFile, stderr=errFile)
         starts.append(time.perf_counter())
         clients.append((popen, cmd))
         files.append((outFile, errFile))
-        Print("Client %d started, Ship node head is: %s" % (i, shipNode.getBlockNum()))
+        Print(f"Client {i} started, Ship node head is: {shipNode.getBlockNum()}")
 
-    Print("Stopping all %d clients" % (args.num_clients))
+    # Generate a fork
+    forkAtProducer="defproducera"
+    prodNode1Prod="defproduceru"
+    preKillBlockNum=nonProdNode.getBlockNum()
+    preKillBlockProducer=nonProdNode.getBlockProducerByNum(preKillBlockNum)
+    nonProdNode.killNodeOnProducer(producer=forkAtProducer, whereInSequence=1)
+    Print(f"Current block producer {preKillBlockProducer} fork will be at producer {forkAtProducer}")
+    prodNode0.waitForProducer(forkAtProducer)
+    prodNode1.waitForProducer(prodNode1Prod)
+    if nonProdNode.verifyAlive(): # if on defproducera, need to wait again
+        prodNode0.waitForProducer(forkAtProducer)
+        prodNode1.waitForProducer(prodNode1Prod)
+    if nonProdNode.verifyAlive():
+        Utils.errorExit("Bridge did not shutdown");
+    Print("Fork started")
 
+    prodNode0.waitForProducer("defproducerb") # wait for fork to progress a bit
+
+    Print("Restore fork")
+    Print("Relaunching the non-producing bridge node to connect the producing nodes again")
+    if nonProdNode.verifyAlive():
+        Utils.errorExit("Bridge is already running");
+    if not nonProdNode.relaunch():
+        Utils.errorExit(f"Failure - (non-production) node {nonProdNode.nodeNum} should have restarted")
+
+    nonProdNode.waitForProducer(forkAtProducer)
+    nonProdNode.waitForProducer(prodNode1Prod)
+    afterForkBlockNum = nonProdNode.getBlockNum()
+    if int(afterForkBlockNum) > int(end_block_num):
+        Utils.errorExit(f"Did not stream long enough {end_block_num} to cover the fork {afterForkBlockNum}, increase block_range {block_range}")
+
+    Print(f"Stopping all {args.num_clients} clients")
     for index, (popen, _), (out, err), start in zip(range(len(clients)), clients, files, starts):
         popen.wait()
-        Print("Stopped client %d.  Ran for %.3f seconds." % (index, time.perf_counter() - start))
+        Print(f"Stopped client {index}.  Ran for {time.perf_counter() - start:.3f} seconds.")
         out.close()
         err.close()
-        outFile = open("%s%d.out" % (shipClientFilePrefix, index), "r")
+        outFile = open(f"{shipClientFilePrefix}{index}.out", "r")
         data = json.load(outFile)
         block_num = start_block_num
         for i in data:
-            print(i)
-            assert block_num == i['get_blocks_result_v0']['this_block']['block_num'], f"{block_num} != {i['get_blocks_result_v0']['this_block']['block_num']}"
+            # fork can cause block numbers to be repeated
+            this_block_num = i['get_blocks_result_v0']['this_block']['block_num']
+            if this_block_num < block_num:
+                block_num = this_block_num
+            assert block_num == this_block_num, f"{block_num} != {this_block_num}"
             assert isinstance(i['get_blocks_result_v0']['block'], str) # verify block in result
             block_num += 1
         assert block_num-1 == end_block_num, f"{block_num-1} != {end_block_num}"

tests/ship_test.py

@@ -132,7 +132,7 @@ try:
             try:
                 statuses = json.loads(" ".join(lines))
             except json.decoder.JSONDecodeError as er:
-                Utils.errorExit("javascript client output was malformed in %s. Exception: %s" % (shipClientErrorFile, er))
+                Utils.errorExit("ship_client output was malformed in %s. Exception: %s" % (shipClientErrorFile, er))
 
             for status in statuses:
                 statusDesc = status["status"]
@@ -143,7 +143,7 @@ try:
                     maxFirstBN = max(maxFirstBN, firstBlockNum)
                     minLastBN = min(minLastBN, lastBlockNum)
                 if statusDesc == "error":
-                    Utils.errorExit("javascript client reporting error see: %s." % (shipClientErrorFile))
+                    Utils.errorExit("ship_client reporting error see: %s." % (shipClientErrorFile))
 
         assert done, Print("ERROR: Did not find a \"done\" status for client %d" % (i))
 

unittests/abi_tests.cpp

@@ -40,7 +40,17 @@ FC_REFLECT(act_sig, (sig) )
 
 BOOST_AUTO_TEST_SUITE(abi_tests)
 
+#ifdef NDEBUG
 fc::microseconds max_serialization_time = fc::seconds(1); // some test machines are very slow
+#else
+fc::microseconds max_serialization_time = fc::microseconds::maximum(); // don't check in debug builds
+#endif
+
+static fc::time_point get_deadline() {
+   if (max_serialization_time == fc::microseconds::maximum())
+      return fc::time_point(fc::microseconds::maximum());
+   return fc::time_point::now() + max_serialization_time;
+}
 
 // verify that round trip conversion, via bytes, reproduces the exact same data
 fc::variant verify_byte_round_trip_conversion( const abi_serializer& abis, const type_name& type, const fc::variant& var )
@@ -49,8 +59,6 @@ fc::variant verify_byte_round_trip_conversion( const abi_serializer& abis, const
 
    auto var2 = abis.binary_to_variant(type, bytes, abi_serializer::create_yield_function( max_serialization_time ));
 
-   std::string r = fc::json::to_string(var2, fc::time_point::now() + max_serialization_time);
-
    auto bytes2 = abis.variant_to_binary(type, var2, abi_serializer::create_yield_function( max_serialization_time ));
 
    BOOST_TEST( fc::to_hex(bytes) == fc::to_hex(bytes2) );
@@ -64,7 +72,7 @@ void verify_round_trip_conversion( const abi_serializer& abis, const type_name&
    auto bytes = abis.variant_to_binary(type, var, abi_serializer::create_yield_function( max_serialization_time ));
    BOOST_REQUIRE_EQUAL(fc::to_hex(bytes), hex);
    auto var2 = abis.binary_to_variant(type, bytes, abi_serializer::create_yield_function( max_serialization_time ));
-   BOOST_REQUIRE_EQUAL(fc::json::to_string(var2, fc::time_point::now() + max_serialization_time), expected_json);
+   BOOST_REQUIRE_EQUAL(fc::json::to_string(var2, get_deadline()), expected_json);
    auto bytes2 = abis.variant_to_binary(type, var2, abi_serializer::create_yield_function( max_serialization_time ));
    BOOST_REQUIRE_EQUAL(fc::to_hex(bytes2), hex);
 }
@@ -94,7 +102,7 @@ fc::variant verify_type_round_trip_conversion( const abi_serializer& abis, const
    fc::variant var2;
    abi_serializer::to_variant(obj, var2, get_resolver(), abi_serializer::create_yield_function( max_serialization_time ));
 
-   std::string r = fc::json::to_string(var2, fc::time_point::now() + max_serialization_time);
+   std::string r = fc::json::to_string(var2, get_deadline());
 
 
    auto bytes2 = abis.variant_to_binary(type, var2, abi_serializer::create_yield_function( max_serialization_time ));
@@ -1929,6 +1937,113 @@ BOOST_AUTO_TEST_CASE(abi_type_loop)
 
 } FC_LOG_AND_RETHROW() }
 
+BOOST_AUTO_TEST_CASE(abi_std_optional)
+{ try {
+   const char* repeat_abi = R"=====(
+   {
+    "version": "eosio::abi/1.2",
+    "types": [],
+    "structs": [
+        {
+            "name": "fees",
+            "base": "",
+            "fields": [
+                {
+                    "name": "gas_price",
+                    "type": "uint64?"
+                },
+                {
+                    "name": "miner_cut",
+                    "type": "uint32?"
+                },
+                {
+                    "name": "bridge_fee",
+                    "type": "uint32?"
+                }
+            ]
+        }
+    ],
+    "actions": [
+        {
+            "name": "fees",
+            "type": "fees",
+            "ricardian_contract": ""
+        }
+    ],
+    "tables": [],
+    "ricardian_clauses": [],
+    "variants": [],
+    "action_results": []
+   }
+   )=====";
+
+   abi_serializer abis(fc::json::from_string(repeat_abi).as<abi_def>(), abi_serializer::create_yield_function( max_serialization_time ));
+   {
+      // check conversion when all optional members are provided
+      std::string test_data = R"=====(
+      {
+        "gas_price" : "42",
+        "miner_cut" : "2",
+        "bridge_fee" : "2"
+      }
+      )=====";
+
+      auto var = fc::json::from_string(test_data);
+      verify_byte_round_trip_conversion(abis, "fees", var);
+   }
+
+   {
+      // check conversion when the first optional member is missing
+      std::string test_data = R"=====(
+      {
+        "miner_cut" : "2",
+        "bridge_fee" : "2"
+      }
+      )=====";
+
+      auto var = fc::json::from_string(test_data);
+      verify_byte_round_trip_conversion(abis, "fees", var);
+   }
+
+   {
+      // check conversion when the second optional member is missing
+      std::string test_data = R"=====(
+      {
+        "gas_price" : "42",
+        "bridge_fee" : "2"
+      }
+      )=====";
+
+      auto var = fc::json::from_string(test_data);
+      verify_byte_round_trip_conversion(abis, "fees", var);
+   }
+
+   {
+      // check conversion when the last optional member is missing
+      std::string test_data = R"=====(
+      {
+        "gas_price" : "42",
+        "miner_cut" : "2",
+      }
+      )=====";
+
+      auto var = fc::json::from_string(test_data);
+      verify_byte_round_trip_conversion(abis, "fees", var);
+   }
+
+   {
+      // check conversion when all optional members are missing
+      std::string test_data = R"=====(
+      {
+      }
+      )=====";
+
+      auto var = fc::json::from_string(test_data);
+      verify_byte_round_trip_conversion(abis, "fees", var);
+   }
+
+} FC_LOG_AND_RETHROW() }
+
 BOOST_AUTO_TEST_CASE(abi_type_redefine)
 { try {
    // inifinite loop in types
@@ -2976,7 +3091,7 @@ BOOST_AUTO_TEST_CASE(abi_to_variant__add_action__good_return_value)
    mutable_variant_object mvo;
    eosio::chain::impl::abi_traverse_context ctx(abi_serializer::create_yield_function(max_serialization_time));
    eosio::chain::impl::abi_to_variant::add(mvo, "action_traces", at, get_resolver(abidef), ctx);
-   std::string res = fc::json::to_string(mvo, fc::time_point::now() + max_serialization_time);
+   std::string res = fc::json::to_string(mvo, get_deadline());
 
    BOOST_CHECK_EQUAL(res, expected_json);
 }
@@ -3001,7 +3116,7 @@ BOOST_AUTO_TEST_CASE(abi_to_variant__add_action__bad_return_value)
    mutable_variant_object mvo;
    eosio::chain::impl::abi_traverse_context ctx(abi_serializer::create_yield_function(max_serialization_time));
    eosio::chain::impl::abi_to_variant::add(mvo, "action_traces", at, get_resolver(abidef), ctx);
-   std::string res = fc::json::to_string(mvo, fc::time_point::now() + max_serialization_time);
+   std::string res = fc::json::to_string(mvo, get_deadline());
 
    BOOST_CHECK_EQUAL(res, expected_json);
 }
@@ -3036,7 +3151,7 @@ BOOST_AUTO_TEST_CASE(abi_to_variant__add_action__no_return_value)
    mutable_variant_object mvo;
    eosio::chain::impl::abi_traverse_context ctx(abi_serializer::create_yield_function(max_serialization_time));
    eosio::chain::impl::abi_to_variant::add(mvo, "action_traces", at, get_resolver(abidef), ctx);
-   std::string res = fc::json::to_string(mvo, fc::time_point::now() + max_serialization_time);
+   std::string res = fc::json::to_string(mvo, get_deadline());
 
    BOOST_CHECK_EQUAL(res, expected_json);
 }

unittests/protocol_feature_tests.cpp

@@ -412,6 +412,14 @@ BOOST_AUTO_TEST_CASE( replace_deferred_test ) try {
    cfg.disable_all_subjective_mitigations = true;
    c.init( cfg );
 
+   transaction_trace_ptr trace;
+   auto h = c.control->applied_transaction.connect( [&](std::tuple<const transaction_trace_ptr&, const packed_transaction_ptr&> x) {
+      auto& t = std::get<0>(x);
+      if( t && !eosio::chain::is_onblock(*t)) {
+         trace = t;
+      }
+   } );
+
    BOOST_CHECK_EQUAL( c.control->get_resource_limits_manager().get_account_ram_usage( "alice"_n ), alice_ram_usage0 );
 
    c.push_action( "test"_n, "defercall"_n, "alice"_n, fc::mutable_variant_object()
@@ -448,6 +456,8 @@ BOOST_AUTO_TEST_CASE( replace_deferred_test ) try {
 
    dtrxs = c.get_scheduled_transactions();
    BOOST_CHECK_EQUAL( dtrxs.size(), 0 );
+   // must be equal before builtin_protocol_feature_t::replace_deferred to support replay of blocks before activation
+   BOOST_CHECK( first_dtrx_id.str() == trace->id.str() );
 
    c.produce_block();
 
@@ -507,6 +517,13 @@ BOOST_AUTO_TEST_CASE( replace_deferred_test ) try {
    BOOST_CHECK_EQUAL( dtrxs.size(), 1 );
    BOOST_CHECK_EQUAL( first_dtrx_id2, dtrxs[0] );
 
+   c.produce_block();
+
+   dtrxs = c.get_scheduled_transactions();
+   BOOST_CHECK_EQUAL( dtrxs.size(), 0 );
+   // Not equal after builtin_protocol_feature_t::replace_deferred activated
+   BOOST_CHECK( first_dtrx_id2.str() != trace->id.str() );
+
 } FC_LOG_AND_RETHROW()
 
 BOOST_AUTO_TEST_CASE( no_duplicate_deferred_id_test ) try {