Merge pull request #953 from AntelopeIO/bump_to_3_2_3

[3.2] Bump Leap version to 3.2.3

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 3)
 #set(VERSION_SUFFIX rc2)
 
 if(VERSION_SUFFIX)

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();

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/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/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;
@@ -693,7 +693,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;
@@ -1607,12 +1607,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 {
+bool producer_plugin_impl::should_interrupt_start_block( const fc::time_point& deadline, uint32_t pending_block_num ) const {
    if( _pending_block_mode == pending_block_mode::producing ) {
       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 +1631,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;
@@ -1696,7 +1697,7 @@ producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
    if (_pending_block_mode == pending_block_mode::producing) {
       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,7 +1711,7 @@ 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;
@@ -1774,7 +1775,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) );
          }
       }
 
@@ -1798,7 +1799,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;
          }
 
@@ -1822,7 +1823,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 +1850,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 ) {
@@ -1885,12 +1887,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;
          }
@@ -2124,6 +2127,8 @@ 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
@@ -2132,7 +2137,7 @@ bool producer_plugin_impl::process_unapplied_trxs( const fc::time_point& deadlin
       auto end_itr = (_pending_block_mode == pending_block_mode::producing) ?
                      _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 +2318,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 +2367,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();
@@ -2565,8 +2571,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);
       }
 

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/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))