diff --git a/libraries/chain/controller.cpp b/libraries/chain/controller.cpp
index 4e9c1439bc..6d8d9a3275 100644
--- a/libraries/chain/controller.cpp
+++ b/libraries/chain/controller.cpp
@@ -2135,44 +2135,65 @@ struct controller_impl {
       }
    } FC_CAPTURE_AND_RETHROW() } /// apply_block
 
+
+   // thread safe, expected to be called from thread other than the main thread
+   block_state_ptr create_block_state_i( const block_id_type& id, const signed_block_ptr& b, const block_header_state& prev ) {
+      auto trx_mroot = calculate_trx_merkle( b->transactions );
+      EOS_ASSERT( b->transaction_mroot == trx_mroot, block_validate_exception,
+                  "invalid block transaction merkle root ${b} != ${c}", ("b", b->transaction_mroot)("c", trx_mroot) );
+
+      const bool skip_validate_signee = false;
+      auto bsp = std::make_shared<block_state>(
+            prev,
+            b,
+            protocol_features.get_protocol_feature_set(),
+            [this]( block_timestamp_type timestamp,
+                    const flat_set<digest_type>& cur_features,
+                    const vector<digest_type>& new_features )
+            { check_protocol_features( timestamp, cur_features, new_features ); },
+            skip_validate_signee
+      );
+
+      EOS_ASSERT( id == bsp->id, block_validate_exception,
+                  "provided id ${id} does not match block id ${bid}", ("id", id)("bid", bsp->id) );
+      return bsp;
+   }
+
    std::future<block_state_ptr> create_block_state_future( const block_id_type& id, const signed_block_ptr& b ) {
       EOS_ASSERT( b, block_validate_exception, "null block" );
 
+      return async_thread_pool( thread_pool.get_executor(), [b, id, control=this]() {
+         // no reason for a block_state if fork_db already knows about block
+         auto existing = control->fork_db.get_block( id );
+         EOS_ASSERT( !existing, fork_database_exception, "we already know about this block: ${id}", ("id", id) );
+
+         auto prev = control->fork_db.get_block_header( b->previous );
+         EOS_ASSERT( prev, unlinkable_block_exception,
+                     "unlinkable block ${id}", ("id", id)("previous", b->previous) );
+
+         return control->create_block_state_i( id, b, *prev );
+      } );
+   }
+
+   // thread safe, expected to be called from thread other than the main thread
+   block_state_ptr create_block_state( const block_id_type& id, const signed_block_ptr& b ) {
+      EOS_ASSERT( b, block_validate_exception, "null block" );
+
       // no reason for a block_state if fork_db already knows about block
       auto existing = fork_db.get_block( id );
       EOS_ASSERT( !existing, fork_database_exception, "we already know about this block: ${id}", ("id", id) );
 
+      // previous not found could mean that previous block not applied yet
       auto prev = fork_db.get_block_header( b->previous );
-      EOS_ASSERT( prev, unlinkable_block_exception,
-                  "unlinkable block ${id}", ("id", id)("previous", b->previous) );
-
-      return async_thread_pool( thread_pool.get_executor(), [b, prev, id, control=this]() {
-         const bool skip_validate_signee = false;
-
-         auto trx_mroot = calculate_trx_merkle( b->transactions );
-         EOS_ASSERT( b->transaction_mroot == trx_mroot, block_validate_exception,
-                     "invalid block transaction merkle root ${b} != ${c}", ("b", b->transaction_mroot)("c", trx_mroot) );
-
-         auto bsp = std::make_shared<block_state>(
-                        *prev,
-                        move( b ),
-                        control->protocol_features.get_protocol_feature_set(),
-                        [control]( block_timestamp_type timestamp,
-                                   const flat_set<digest_type>& cur_features,
-                                   const vector<digest_type>& new_features )
-                        { control->check_protocol_features( timestamp, cur_features, new_features ); },
-                        skip_validate_signee
-         );
+      if( !prev ) return {};
 
-         EOS_ASSERT( id == bsp->id, block_validate_exception,
-                     "provided id ${id} does not match block id ${bid}", ("id", id)("bid", bsp->id) );
-         return bsp;
-      } );
+      return create_block_state_i( id, b, *prev );
    }
 
    void push_block( controller::block_report& br,
-                    std::future<block_state_ptr>& block_state_future,
-                    const forked_branch_callback& forked_branch_cb, const trx_meta_cache_lookup& trx_lookup )
+                    const block_state_ptr& bsp,
+                    const forked_branch_callback& forked_branch_cb,
+                    const trx_meta_cache_lookup& trx_lookup )
    {
       controller::block_status s = controller::block_status::complete;
       EOS_ASSERT(!pending, block_validate_exception, "it is not valid to push a block when there is a pending block");
@@ -2181,7 +2202,7 @@ struct controller_impl {
          trusted_producer_light_validation = old_value;
       });
       try {
-         block_state_ptr bsp = block_state_future.get();
+         EOS_ASSERT( bsp, block_validate_exception, "null block" );
          const auto& b = bsp->block;
 
          if( conf.terminate_at_block > 0 && conf.terminate_at_block < self.head_block_num()) {
@@ -2900,12 +2921,17 @@ std::future<block_state_ptr> controller::create_block_state_future( const block_
    return my->create_block_state_future( id, b );
 }
 
+block_state_ptr controller::create_block_state( const block_id_type& id, const signed_block_ptr& b ) const {
+   return my->create_block_state( id, b );
+}
+
 void controller::push_block( controller::block_report& br,
-                             std::future<block_state_ptr>& block_state_future,
-                             const forked_branch_callback& forked_branch_cb, const trx_meta_cache_lookup& trx_lookup )
+                             const block_state_ptr& bsp,
+                             const forked_branch_callback& forked_branch_cb,
+                             const trx_meta_cache_lookup& trx_lookup )
 {
    validate_db_available_size();
-   my->push_block( br, block_state_future, forked_branch_cb, trx_lookup );
+   my->push_block( br, bsp, forked_branch_cb, trx_lookup );
 }
 
 transaction_trace_ptr controller::push_transaction( const transaction_metadata_ptr& trx,
diff --git a/libraries/chain/fork_database.cpp b/libraries/chain/fork_database.cpp
index 05af8746be..8d8a6d8495 100644
--- a/libraries/chain/fork_database.cpp
+++ b/libraries/chain/fork_database.cpp
@@ -9,6 +9,7 @@
 #include <boost/multi_index/composite_key.hpp>
 #include <fc/io/fstream.hpp>
 #include <fstream>
+#include <shared_mutex>
 
 namespace eosio { namespace chain {
    using boost::multi_index_container;
@@ -60,27 +61,44 @@ namespace eosio { namespace chain {
    }
 
    struct fork_database_impl {
-      fork_database_impl( fork_database& self, const fc::path& data_dir )
-      :self(self)
-      ,datadir(data_dir)
+      explicit fork_database_impl( const fc::path& data_dir )
+      :datadir(data_dir)
       {}
 
-      fork_database&        self;
+      std::shared_mutex     mtx;
       fork_multi_index_type index;
       block_state_ptr       root; // Only uses the block_header_state portion
       block_state_ptr       head;
       fc::path              datadir;
 
-      void add( const block_state_ptr& n,
-                bool ignore_duplicate, bool validate,
-                const std::function<void( block_timestamp_type,
-                                          const flat_set<digest_type>&,
-                                          const vector<digest_type>& )>& validator );
+      void open_impl( const std::function<void( block_timestamp_type,
+                                                const flat_set<digest_type>&,
+                                                const vector<digest_type>& )>& validator );
+      void close_impl();
+
+
+      block_header_state_ptr  get_block_header_impl( const block_id_type& id )const;
+      block_state_ptr         get_block_impl( const block_id_type& id )const;
+      void            reset_impl( const block_header_state& root_bhs );
+      void            rollback_head_to_root_impl();
+      void            advance_root_impl( const block_id_type& id );
+      void            remove_impl( const block_id_type& id );
+      branch_type     fetch_branch_impl( const block_id_type& h, uint32_t trim_after_block_num )const;
+      block_state_ptr search_on_branch_impl( const block_id_type& h, uint32_t block_num )const;
+      pair<branch_type, branch_type> fetch_branch_from_impl( const block_id_type& first,
+                                                             const block_id_type& second )const;
+      void mark_valid_impl( const block_state_ptr& h );
+
+      void add_impl( const block_state_ptr& n,
+                     bool ignore_duplicate, bool validate,
+                     const std::function<void( block_timestamp_type,
+                                               const flat_set<digest_type>&,
+                                               const vector<digest_type>& )>& validator );
    };
 
 
    fork_database::fork_database( const fc::path& data_dir )
-   :my( new fork_database_impl( *this, data_dir ) )
+   :my( new fork_database_impl( data_dir ) )
    {}
 
 
@@ -88,10 +106,18 @@ namespace eosio { namespace chain {
                                                        const flat_set<digest_type>&,
                                                        const vector<digest_type>& )>& validator )
    {
-      if (!fc::is_directory(my->datadir))
-         fc::create_directories(my->datadir);
+      std::lock_guard g( my->mtx );
+      my->open_impl( validator );
+   }
+
+   void fork_database_impl::open_impl( const std::function<void( block_timestamp_type,
+                                                                 const flat_set<digest_type>&,
+                                                                 const vector<digest_type>& )>& validator )
+   {
+      if (!fc::is_directory(datadir))
+         fc::create_directories(datadir);
 
-      auto fork_db_dat = my->datadir / config::forkdb_filename;
+      auto fork_db_dat = datadir / config::forkdb_filename;
       if( fc::exists( fork_db_dat ) ) {
          try {
             string content;
@@ -102,29 +128,29 @@ namespace eosio { namespace chain {
             // validate totem
             uint32_t totem = 0;
             fc::raw::unpack( ds, totem );
-            EOS_ASSERT( totem == magic_number, fork_database_exception,
+            EOS_ASSERT( totem == fork_database::magic_number, fork_database_exception,
                         "Fork database file '${filename}' has unexpected magic number: ${actual_totem}. Expected ${expected_totem}",
                         ("filename", fork_db_dat.generic_string())
                         ("actual_totem", totem)
-                        ("expected_totem", magic_number)
+                        ("expected_totem", fork_database::magic_number)
             );
 
             // validate version
             uint32_t version = 0;
             fc::raw::unpack( ds, version );
-            EOS_ASSERT( version >= min_supported_version && version <= max_supported_version,
+            EOS_ASSERT( version >= fork_database::min_supported_version && version <= fork_database::max_supported_version,
                         fork_database_exception,
                        "Unsupported version of fork database file '${filename}'. "
                        "Fork database version is ${version} while code supports version(s) [${min},${max}]",
                        ("filename", fork_db_dat.generic_string())
                        ("version", version)
-                       ("min", min_supported_version)
-                       ("max", max_supported_version)
+                       ("min", fork_database::min_supported_version)
+                       ("max", fork_database::max_supported_version)
             );
 
             block_header_state bhs;
             fc::raw::unpack( ds, bhs );
-            reset( bhs );
+            reset_impl( bhs );
 
             unsigned_int size; fc::raw::unpack( ds, size );
             for( uint32_t i = 0, n = size.value; i < n; ++i ) {
@@ -132,27 +158,27 @@ namespace eosio { namespace chain {
                fc::raw::unpack( ds, s );
                // do not populate transaction_metadatas, they will be created as needed in apply_block with appropriate key recovery
                s.header_exts = s.block->validate_and_extract_header_extensions();
-               my->add( std::make_shared<block_state>( move( s ) ), false, true, validator );
+               add_impl( std::make_shared<block_state>( std::move( s ) ), false, true, validator );
             }
             block_id_type head_id;
             fc::raw::unpack( ds, head_id );
 
-            if( my->root->id == head_id ) {
-               my->head = my->root;
+            if( root->id == head_id ) {
+               head = root;
             } else {
-               my->head = get_block( head_id );
-               EOS_ASSERT( my->head, fork_database_exception,
+               head = get_block_impl( head_id );
+               EOS_ASSERT( head, fork_database_exception,
                            "could not find head while reconstructing fork database from file; '${filename}' is likely corrupted",
                            ("filename", fork_db_dat.generic_string()) );
             }
 
-            auto candidate = my->index.get<by_lib_block_num>().begin();
-            if( candidate == my->index.get<by_lib_block_num>().end() || !(*candidate)->is_valid() ) {
-               EOS_ASSERT( my->head->id == my->root->id, fork_database_exception,
+            auto candidate = index.get<by_lib_block_num>().begin();
+            if( candidate == index.get<by_lib_block_num>().end() || !(*candidate)->is_valid() ) {
+               EOS_ASSERT( head->id == root->id, fork_database_exception,
                            "head not set to root despite no better option available; '${filename}' is likely corrupted",
                            ("filename", fork_db_dat.generic_string()) );
             } else {
-               EOS_ASSERT( !first_preferred( **candidate, *my->head ), fork_database_exception,
+               EOS_ASSERT( !first_preferred( **candidate, *head ), fork_database_exception,
                            "head not set to best available option available; '${filename}' is likely corrupted",
                            ("filename", fork_db_dat.generic_string()) );
             }
@@ -163,10 +189,15 @@ namespace eosio { namespace chain {
    }
 
    void fork_database::close() {
-      auto fork_db_dat = my->datadir / config::forkdb_filename;
+      std::lock_guard g( my->mtx );
+      my->close_impl();
+   }
 
-      if( !my->root ) {
-         if( my->index.size() > 0 ) {
+   void fork_database_impl::close_impl() {
+      auto fork_db_dat = datadir / config::forkdb_filename;
+
+      if( !root ) {
+         if( index.size() > 0 ) {
             elog( "fork_database is in a bad state when closing; not writing out '${filename}'",
                   ("filename", fork_db_dat.generic_string()) );
          }
@@ -174,13 +205,13 @@ namespace eosio { namespace chain {
       }
 
       std::ofstream out( fork_db_dat.generic_string().c_str(), std::ios::out | std::ios::binary | std::ofstream::trunc );
-      fc::raw::pack( out, magic_number );
-      fc::raw::pack( out, max_supported_version ); // write out current version which is always max_supported_version
-      fc::raw::pack( out, *static_cast<block_header_state*>(&*my->root) );
-      uint32_t num_blocks_in_fork_db = my->index.size();
+      fc::raw::pack( out, fork_database::magic_number );
+      fc::raw::pack( out, fork_database::max_supported_version ); // write out current version which is always max_supported_version
+      fc::raw::pack( out, *static_cast<block_header_state*>(&*root) );
+      uint32_t num_blocks_in_fork_db = index.size();
       fc::raw::pack( out, unsigned_int{num_blocks_in_fork_db} );
 
-      const auto& indx = my->index.get<by_lib_block_num>();
+      const auto& indx = index.get<by_lib_block_num>();
 
       auto unvalidated_itr = indx.rbegin();
       auto unvalidated_end = boost::make_reverse_iterator( indx.lower_bound( false ) );
@@ -215,30 +246,40 @@ namespace eosio { namespace chain {
          fc::raw::pack( out, *(*itr) );
       }
 
-      if( my->head ) {
-         fc::raw::pack( out, my->head->id );
+      if( head ) {
+         fc::raw::pack( out, head->id );
       } else {
          elog( "head not set in fork database; '${filename}' will be corrupted",
                ("filename", fork_db_dat.generic_string()) );
       }
 
-      my->index.clear();
+      index.clear();
    }
 
    fork_database::~fork_database() {
-      close();
+      my->close_impl();
    }
 
    void fork_database::reset( const block_header_state& root_bhs ) {
-      my->index.clear();
-      my->root = std::make_shared<block_state>();
-      static_cast<block_header_state&>(*my->root) = root_bhs;
-      my->root->validated = true;
-      my->head = my->root;
+      std::lock_guard g( my->mtx );
+      my->reset_impl(root_bhs);
+   }
+
+   void fork_database_impl::reset_impl( const block_header_state& root_bhs ) {
+      index.clear();
+      root = std::make_shared<block_state>();
+      static_cast<block_header_state&>(*root) = root_bhs;
+      root->validated = true;
+      head = root;
    }
 
    void fork_database::rollback_head_to_root() {
-      auto& by_id_idx = my->index.get<by_block_id>();
+      std::lock_guard g( my->mtx );
+      my->rollback_head_to_root_impl();
+   }
+
+   void fork_database_impl::rollback_head_to_root_impl() {
+      auto& by_id_idx = index.get<by_block_id>();
       auto itr = by_id_idx.begin();
       while (itr != by_id_idx.end()) {
          by_id_idx.modify( itr, [&]( block_state_ptr& bsp ) {
@@ -246,13 +287,18 @@ namespace eosio { namespace chain {
          } );
          ++itr;
       }
-      my->head = my->root;
+      head = root;
    }
 
    void fork_database::advance_root( const block_id_type& id ) {
-      EOS_ASSERT( my->root, fork_database_exception, "root not yet set" );
+      std::lock_guard g( my->mtx );
+      my->advance_root_impl( id );
+   }
+
+   void fork_database_impl::advance_root_impl( const block_id_type& id ) {
+      EOS_ASSERT( root, fork_database_exception, "root not yet set" );
 
-      auto new_root = get_block( id );
+      auto new_root = get_block_impl( id );
       EOS_ASSERT( new_root, fork_database_exception,
                   "cannot advance root to a block that does not exist in the fork database" );
       EOS_ASSERT( new_root->is_valid(), fork_database_exception,
@@ -262,48 +308,53 @@ namespace eosio { namespace chain {
       deque<block_id_type> blocks_to_remove;
       for( auto b = new_root; b; ) {
          blocks_to_remove.emplace_back( b->header.previous );
-         b = get_block( blocks_to_remove.back() );
-         EOS_ASSERT( b || blocks_to_remove.back() == my->root->id, fork_database_exception, "invariant violation: orphaned branch was present in forked database" );
+         b = get_block_impl( blocks_to_remove.back() );
+         EOS_ASSERT( b || blocks_to_remove.back() == root->id, fork_database_exception, "invariant violation: orphaned branch was present in forked database" );
       }
 
       // The new root block should be erased from the fork database index individually rather than with the remove method,
       // because we do not want the blocks branching off of it to be removed from the fork database.
-      my->index.erase( my->index.find( id ) );
+      index.erase( index.find( id ) );
 
       // The other blocks to be removed are removed using the remove method so that orphaned branches do not remain in the fork database.
       for( const auto& block_id : blocks_to_remove ) {
-         remove( block_id );
+         remove_impl( block_id );
       }
 
       // Even though fork database no longer needs block or trxs when a block state becomes a root of the tree,
       // avoid mutating the block state at all, for example clearing the block shared pointer, because other
       // parts of the code which run asynchronously may later expect it remain unmodified.
 
-      my->root = new_root;
+      root = new_root;
    }
 
    block_header_state_ptr fork_database::get_block_header( const block_id_type& id )const {
-      if( my->root->id == id ) {
-         return my->root;
+      std::shared_lock g( my->mtx );
+      return my->get_block_header_impl( id );
+   }
+
+   block_header_state_ptr fork_database_impl::get_block_header_impl( const block_id_type& id )const {
+      if( root->id == id ) {
+         return root;
       }
 
-      auto itr = my->index.find( id );
-      if( itr != my->index.end() )
+      auto itr = index.find( id );
+      if( itr != index.end() )
          return *itr;
 
       return block_header_state_ptr();
    }
 
-   void fork_database_impl::add( const block_state_ptr& n,
-                                 bool ignore_duplicate, bool validate,
-                                 const std::function<void( block_timestamp_type,
-                                                           const flat_set<digest_type>&,
-                                                           const vector<digest_type>& )>& validator )
+   void fork_database_impl::add_impl( const block_state_ptr& n,
+                                      bool ignore_duplicate, bool validate,
+                                      const std::function<void( block_timestamp_type,
+                                                                const flat_set<digest_type>&,
+                                                                const vector<digest_type>& )>& validator )
    {
       EOS_ASSERT( root, fork_database_exception, "root not yet set" );
       EOS_ASSERT( n, fork_database_exception, "attempt to add null block state" );
 
-      auto prev_bh = self.get_block_header( n->header.previous );
+      auto prev_bh = get_block_header_impl( n->header.previous );
 
       EOS_ASSERT( prev_bh, unlinkable_block_exception,
                   "unlinkable block", ("id", n->id)("previous", n->header.previous) );
@@ -332,19 +383,27 @@ namespace eosio { namespace chain {
    }
 
    void fork_database::add( const block_state_ptr& n, bool ignore_duplicate ) {
-      my->add( n, ignore_duplicate, false,
-               []( block_timestamp_type timestamp,
-                   const flat_set<digest_type>& cur_features,
-                   const vector<digest_type>& new_features )
-               {}
+      std::lock_guard g( my->mtx );
+      my->add_impl( n, ignore_duplicate, false,
+                    []( block_timestamp_type timestamp,
+                        const flat_set<digest_type>& cur_features,
+                        const vector<digest_type>& new_features )
+                    {}
       );
    }
 
-   const block_state_ptr& fork_database::root()const { return my->root; }
+   block_state_ptr fork_database::root()const {
+      std::shared_lock g( my->mtx );
+      return my->root;
+   }
 
-   const block_state_ptr& fork_database::head()const { return my->head; }
+   block_state_ptr fork_database::head()const {
+      std::shared_lock g( my->mtx );
+      return my->head;
+   }
 
    block_state_ptr fork_database::pending_head()const {
+      std::shared_lock g( my->mtx );
       const auto& indx = my->index.get<by_lib_block_num>();
 
       auto itr = indx.lower_bound( false );
@@ -357,8 +416,13 @@ namespace eosio { namespace chain {
    }
 
    branch_type fork_database::fetch_branch( const block_id_type& h, uint32_t trim_after_block_num )const {
+      std::shared_lock g( my->mtx );
+      return my->fetch_branch_impl( h, trim_after_block_num );
+   }
+
+   branch_type fork_database_impl::fetch_branch_impl( const block_id_type& h, uint32_t trim_after_block_num )const {
       branch_type result;
-      for( auto s = get_block(h); s; s = get_block( s->header.previous ) ) {
+      for( auto s = get_block_impl(h); s; s = get_block_impl( s->header.previous ) ) {
          if( s->block_num <= trim_after_block_num )
              result.push_back( s );
       }
@@ -367,7 +431,12 @@ namespace eosio { namespace chain {
    }
 
    block_state_ptr fork_database::search_on_branch( const block_id_type& h, uint32_t block_num )const {
-      for( auto s = get_block(h); s; s = get_block( s->header.previous ) ) {
+      std::shared_lock g( my->mtx );
+      return my->search_on_branch_impl( h, block_num );
+   }
+
+   block_state_ptr fork_database_impl::search_on_branch_impl( const block_id_type& h, uint32_t block_num )const {
+      for( auto s = get_block_impl(h); s; s = get_block_impl( s->header.previous ) ) {
          if( s->block_num == block_num )
              return s;
       }
@@ -381,9 +450,15 @@ namespace eosio { namespace chain {
     */
    pair< branch_type, branch_type >  fork_database::fetch_branch_from( const block_id_type& first,
                                                                        const block_id_type& second )const {
+      std::shared_lock g( my->mtx );
+      return my->fetch_branch_from_impl( first, second );
+   }
+
+   pair< branch_type, branch_type >  fork_database_impl::fetch_branch_from_impl( const block_id_type& first,
+                                                                                 const block_id_type& second )const {
       pair<branch_type,branch_type> result;
-      auto first_branch = (first == my->root->id) ? my->root : get_block(first);
-      auto second_branch = (second == my->root->id) ? my->root : get_block(second);
+      auto first_branch = (first == root->id) ? root : get_block_impl(first);
+      auto second_branch = (second == root->id) ? root : get_block_impl(second);
 
       EOS_ASSERT(first_branch, fork_db_block_not_found, "block ${id} does not exist", ("id", first));
       EOS_ASSERT(second_branch, fork_db_block_not_found, "block ${id} does not exist", ("id", second));
@@ -392,7 +467,7 @@ namespace eosio { namespace chain {
       {
          result.first.push_back(first_branch);
          const auto& prev = first_branch->header.previous;
-         first_branch = (prev == my->root->id) ? my->root : get_block( prev );
+         first_branch = (prev == root->id) ? root : get_block_impl( prev );
          EOS_ASSERT( first_branch, fork_db_block_not_found,
                      "block ${id} does not exist",
                      ("id", prev)
@@ -403,7 +478,7 @@ namespace eosio { namespace chain {
       {
          result.second.push_back( second_branch );
          const auto& prev = second_branch->header.previous;
-         second_branch = (prev == my->root->id) ? my->root : get_block( prev );
+         second_branch = (prev == root->id) ? root : get_block_impl( prev );
          EOS_ASSERT( second_branch, fork_db_block_not_found,
                      "block ${id} does not exist",
                      ("id", prev)
@@ -417,9 +492,9 @@ namespace eosio { namespace chain {
          result.first.push_back(first_branch);
          result.second.push_back(second_branch);
          const auto &first_prev = first_branch->header.previous;
-         first_branch = get_block( first_prev );
+         first_branch = get_block_impl( first_prev );
          const auto &second_prev = second_branch->header.previous;
-         second_branch = get_block( second_prev );
+         second_branch = get_block_impl( second_prev );
          EOS_ASSERT( first_branch, fork_db_block_not_found,
                      "block ${id} does not exist",
                      ("id", first_prev)
@@ -436,13 +511,18 @@ namespace eosio { namespace chain {
          result.second.push_back(second_branch);
       }
       return result;
-   } /// fetch_branch_from
+   } /// fetch_branch_from_impl
 
    /// remove all of the invalid forks built off of this id including this id
    void fork_database::remove( const block_id_type& id ) {
+      std::lock_guard g( my->mtx );
+      return my->remove_impl( id );
+   }
+
+   void fork_database_impl::remove_impl( const block_id_type& id ) {
       deque<block_id_type> remove_queue{id};
-      const auto& previdx = my->index.get<by_prev>();
-      const auto& head_id = my->head->id;
+      const auto& previdx = index.get<by_prev>();
+      const auto& head_id = head->id;
 
       for( uint32_t i = 0; i < remove_queue.size(); ++i ) {
          EOS_ASSERT( remove_queue[i] != head_id, fork_database_exception,
@@ -456,16 +536,19 @@ namespace eosio { namespace chain {
       }
 
       for( const auto& block_id : remove_queue ) {
-         auto itr = my->index.find( block_id );
-         if( itr != my->index.end() )
-            my->index.erase(itr);
+         index.erase( block_id );
       }
    }
 
    void fork_database::mark_valid( const block_state_ptr& h ) {
+      std::lock_guard g( my->mtx );
+      my->mark_valid_impl( h );
+   }
+
+   void fork_database_impl::mark_valid_impl( const block_state_ptr& h ) {
       if( h->validated ) return;
 
-      auto& by_id_idx = my->index.get<by_block_id>();
+      auto& by_id_idx = index.get<by_block_id>();
 
       auto itr = by_id_idx.find( h->id );
       EOS_ASSERT( itr != by_id_idx.end(), fork_database_exception,
@@ -476,15 +559,20 @@ namespace eosio { namespace chain {
          bsp->validated = true;
       } );
 
-      auto candidate = my->index.get<by_lib_block_num>().begin();
-      if( first_preferred( **candidate, *my->head ) ) {
-         my->head = *candidate;
+      auto candidate = index.get<by_lib_block_num>().begin();
+      if( first_preferred( **candidate, *head ) ) {
+         head = *candidate;
       }
    }
 
-   block_state_ptr   fork_database::get_block(const block_id_type& id)const {
-      auto itr = my->index.find( id );
-      if( itr != my->index.end() )
+   block_state_ptr fork_database::get_block(const block_id_type& id)const {
+      std::shared_lock g( my->mtx );
+      return my->get_block_impl(id);
+   }
+
+   block_state_ptr fork_database_impl::get_block_impl(const block_id_type& id)const {
+      auto itr = index.find( id );
+      if( itr != index.end() )
          return *itr;
       return block_state_ptr();
    }
diff --git a/libraries/chain/include/eosio/chain/controller.hpp b/libraries/chain/include/eosio/chain/controller.hpp
index da6fe4c4d1..37b6eabe02 100644
--- a/libraries/chain/include/eosio/chain/controller.hpp
+++ b/libraries/chain/include/eosio/chain/controller.hpp
@@ -175,16 +175,19 @@ namespace eosio { namespace chain {
          void sign_block( const signer_callback_type& signer_callback );
          void commit_block();
 
+         // thread-safe
          std::future<block_state_ptr> create_block_state_future( const block_id_type& id, const signed_block_ptr& b );
+         // thread-safe
+         block_state_ptr create_block_state( const block_id_type& id, const signed_block_ptr& b ) const;
 
          /**
           * @param br returns statistics for block
-          * @param block_state_future provide from call to create_block_state_future
+          * @param bsp block to push
           * @param cb calls cb with forked applied transactions for each forked block
           * @param trx_lookup user provided lookup function for externally cached transaction_metadata
           */
          void push_block( block_report& br,
-                          std::future<block_state_ptr>& block_state_future,
+                          const block_state_ptr& bsp,
                           const forked_branch_callback& cb,
                           const trx_meta_cache_lookup& trx_lookup );
 
@@ -253,6 +256,7 @@ namespace eosio { namespace chain {
          signed_block_ptr fetch_block_by_id( block_id_type id )const;
 
          block_state_ptr fetch_block_state_by_number( uint32_t block_num )const;
+         // return block_state from forkdb, thread-safe
          block_state_ptr fetch_block_state_by_id( block_id_type id )const;
 
          block_id_type get_block_id_for_num( uint32_t block_num )const;
diff --git a/libraries/chain/include/eosio/chain/fork_database.hpp b/libraries/chain/include/eosio/chain/fork_database.hpp
index bd774f2f71..bc9f119912 100644
--- a/libraries/chain/include/eosio/chain/fork_database.hpp
+++ b/libraries/chain/include/eosio/chain/fork_database.hpp
@@ -16,6 +16,8 @@ namespace eosio { namespace chain {
     * database tracks the longest chain and the last irreversible block number. All
     * blocks older than the last irreversible block are freed after emitting the
     * irreversible signal.
+    *
+    * An internal mutex is used to provide thread-safety.
     */
    class fork_database {
       public:
@@ -55,9 +57,9 @@ namespace eosio { namespace chain {
 
          void            remove( const block_id_type& id );
 
-         const block_state_ptr& root()const;
-         const block_state_ptr& head()const;
-         block_state_ptr        pending_head()const;
+         block_state_ptr root()const;
+         block_state_ptr head()const;
+         block_state_ptr pending_head()const;
 
          /**
           *  Returns the sequence of block states resulting from trimming the branch from the
diff --git a/libraries/chain/include/eosio/chain/unapplied_transaction_queue.hpp b/libraries/chain/include/eosio/chain/unapplied_transaction_queue.hpp
index 5020b08b28..43166e9371 100644
--- a/libraries/chain/include/eosio/chain/unapplied_transaction_queue.hpp
+++ b/libraries/chain/include/eosio/chain/unapplied_transaction_queue.hpp
@@ -96,15 +96,15 @@ class unapplied_transaction_queue {
       return itr->trx_meta;
    }
 
-   template <typename Func>
-   bool clear_expired( const time_point& pending_block_time, const time_point& deadline, Func&& callback ) {
+   template <typename Yield, typename Callback>
+   bool clear_expired( const time_point& pending_block_time, Yield&& yield, Callback&& callback ) {
       auto& persisted_by_expiry = queue.get<by_expiry>();
       while( !persisted_by_expiry.empty() ) {
          const auto& itr = persisted_by_expiry.begin();
          if( itr->expiration() > pending_block_time ) {
             break;
          }
-         if( deadline <= fc::time_point::now() ) {
+         if( yield() ) {
             return false;
          }
          callback( itr->trx_meta->packed_trx(), itr->trx_type );
diff --git a/libraries/testing/include/eosio/testing/tester.hpp b/libraries/testing/include/eosio/testing/tester.hpp
index d3069649a6..af2c4a8d31 100644
--- a/libraries/testing/include/eosio/testing/tester.hpp
+++ b/libraries/testing/include/eosio/testing/tester.hpp
@@ -597,7 +597,7 @@ namespace eosio { namespace testing {
          auto sb = _produce_block(skip_time, false);
          auto bsf = validating_node->create_block_state_future( sb->calculate_id(), sb );
          controller::block_report br;
-         validating_node->push_block( br, bsf, forked_branch_callback{}, trx_meta_cache_lookup{} );
+         validating_node->push_block( br, bsf.get(), forked_branch_callback{}, trx_meta_cache_lookup{} );
 
          return sb;
       }
@@ -607,9 +607,9 @@ namespace eosio { namespace testing {
       }
 
       void validate_push_block(const signed_block_ptr& sb) {
-         auto bs = validating_node->create_block_state_future( sb->calculate_id(), sb );
+         auto bsf = validating_node->create_block_state_future( sb->calculate_id(), sb );
          controller::block_report br;
-         validating_node->push_block( br, bs, forked_branch_callback{}, trx_meta_cache_lookup{} );
+         validating_node->push_block( br, bsf.get(), forked_branch_callback{}, trx_meta_cache_lookup{} );
       }
 
       signed_block_ptr produce_empty_block( fc::microseconds skip_time = fc::milliseconds(config::block_interval_ms) )override {
@@ -617,7 +617,7 @@ namespace eosio { namespace testing {
          auto sb = _produce_block(skip_time, true);
          auto bsf = validating_node->create_block_state_future( sb->calculate_id(), sb );
          controller::block_report br;
-         validating_node->push_block( br, bsf, forked_branch_callback{}, trx_meta_cache_lookup{} );
+         validating_node->push_block( br, bsf.get(), forked_branch_callback{}, trx_meta_cache_lookup{} );
 
          return sb;
       }
diff --git a/libraries/testing/tester.cpp b/libraries/testing/tester.cpp
index 932de3b88f..314a16072a 100644
--- a/libraries/testing/tester.cpp
+++ b/libraries/testing/tester.cpp
@@ -322,7 +322,7 @@ namespace eosio { namespace testing {
       auto bsf = control->create_block_state_future(b->calculate_id(), b);
       unapplied_transactions.add_aborted( control->abort_block() );
       controller::block_report br;
-      control->push_block( br, bsf, [this]( const branch_type& forked_branch ) {
+      control->push_block( br, bsf.get(), [this]( const branch_type& forked_branch ) {
          unapplied_transactions.add_forked( forked_branch );
       }, [this]( const transaction_id_type& id ) {
          return unapplied_transactions.get_trx( id );
@@ -1052,7 +1052,7 @@ namespace eosio { namespace testing {
                auto bsf = b.control->create_block_state_future( block->calculate_id(), block );
                b.control->abort_block();
                controller::block_report br;
-               b.control->push_block(br, bsf, forked_branch_callback{}, trx_meta_cache_lookup{}); //, eosio::chain::validation_steps::created_block);
+               b.control->push_block(br, bsf.get(), forked_branch_callback{}, trx_meta_cache_lookup{}); //, eosio::chain::validation_steps::created_block);
             }
          }
       };
diff --git a/plugins/chain_interface/include/eosio/chain/plugin_interface.hpp b/plugins/chain_interface/include/eosio/chain/plugin_interface.hpp
index 6a02750d62..6e109d63a9 100644
--- a/plugins/chain_interface/include/eosio/chain/plugin_interface.hpp
+++ b/plugins/chain_interface/include/eosio/chain/plugin_interface.hpp
@@ -37,14 +37,9 @@ namespace eosio { namespace chain { namespace plugin_interface {
    }
 
    namespace incoming {
-      namespace channels {
-         using block                 = channel_decl<struct block_tag, signed_block_ptr>;
-         using transaction           = channel_decl<struct transaction_tag, packed_transaction_ptr>;
-      }
-
       namespace methods {
-         // synchronously push a block/trx to a single provider
-         using block_sync            = method_decl<chain_plugin_interface, bool(const signed_block_ptr&, const std::optional<block_id_type>&), first_provider_policy>;
+         // synchronously push a block/trx to a single provider, block_state_ptr may be null
+         using block_sync            = method_decl<chain_plugin_interface, bool(const signed_block_ptr&, const std::optional<block_id_type>&, const block_state_ptr&), first_provider_policy>;
          using transaction_async     = method_decl<chain_plugin_interface, void(const packed_transaction_ptr&, bool, bool, bool, next_function<transaction_trace_ptr>), first_provider_policy>;
       }
    }
diff --git a/plugins/chain_plugin/chain_plugin.cpp b/plugins/chain_plugin/chain_plugin.cpp
index 03d0f6128e..f2cf67de2b 100644
--- a/plugins/chain_plugin/chain_plugin.cpp
+++ b/plugins/chain_plugin/chain_plugin.cpp
@@ -142,7 +142,6 @@ class chain_plugin_impl {
    ,irreversible_block_channel(app().get_channel<channels::irreversible_block>())
    ,accepted_transaction_channel(app().get_channel<channels::accepted_transaction>())
    ,applied_transaction_channel(app().get_channel<channels::applied_transaction>())
-   ,incoming_block_channel(app().get_channel<incoming::channels::block>())
    ,incoming_block_sync_method(app().get_method<incoming::methods::block_sync>())
    ,incoming_transaction_async_method(app().get_method<incoming::methods::transaction_async>())
    {}
@@ -170,7 +169,6 @@ class chain_plugin_impl {
    channels::irreversible_block::channel_type&     irreversible_block_channel;
    channels::accepted_transaction::channel_type&   accepted_transaction_channel;
    channels::applied_transaction::channel_type&    applied_transaction_channel;
-   incoming::channels::block::channel_type&        incoming_block_channel;
 
    // retained references to methods for easy calling
    incoming::methods::block_sync::method_type&        incoming_block_sync_method;
@@ -1352,8 +1350,8 @@ chain_apis::read_only chain_plugin::get_read_only_api(const fc::microseconds& ht
 }
 
 
-bool chain_plugin::accept_block(const signed_block_ptr& block, const block_id_type& id ) {
-   return my->incoming_block_sync_method(block, id);
+bool chain_plugin::accept_block(const signed_block_ptr& block, const block_id_type& id, const block_state_ptr& bsp ) {
+   return my->incoming_block_sync_method(block, id, bsp);
 }
 
 void chain_plugin::accept_transaction(const chain::packed_transaction_ptr& trx, next_function<chain::transaction_trace_ptr> next) {
@@ -2172,7 +2170,7 @@ fc::variant read_only::get_block_header_state(const get_block_header_state_param
 
 void read_write::push_block(read_write::push_block_params&& params, next_function<read_write::push_block_results> next) {
    try {
-      app().get_method<incoming::methods::block_sync>()(std::make_shared<signed_block>( std::move( params ) ), std::optional<block_id_type>{});
+      app().get_method<incoming::methods::block_sync>()(std::make_shared<signed_block>( std::move(params) ), std::optional<block_id_type>{}, block_state_ptr{});
    } catch ( boost::interprocess::bad_alloc& ) {
       chain_plugin::handle_db_exhaustion();
    } catch ( const std::bad_alloc& ) {
diff --git a/plugins/chain_plugin/include/eosio/chain_plugin/chain_plugin.hpp b/plugins/chain_plugin/include/eosio/chain_plugin/chain_plugin.hpp
index cdf1ae97a7..094c1e28eb 100644
--- a/plugins/chain_plugin/include/eosio/chain_plugin/chain_plugin.hpp
+++ b/plugins/chain_plugin/include/eosio/chain_plugin/chain_plugin.hpp
@@ -849,7 +849,7 @@ class chain_plugin : public plugin<chain_plugin> {
    chain_apis::read_write get_read_write_api(const fc::microseconds& http_max_response_time);
    chain_apis::read_only get_read_only_api(const fc::microseconds& http_max_response_time) const;
 
-   bool accept_block( const chain::signed_block_ptr& block, const chain::block_id_type& id );
+   bool accept_block( const chain::signed_block_ptr& block, const chain::block_id_type& id, const chain::block_state_ptr& bsp );
    void accept_transaction(const chain::packed_transaction_ptr& trx, chain::plugin_interface::next_function<chain::transaction_trace_ptr> next);
 
    // Only call this after plugin_initialize()!
diff --git a/plugins/net_plugin/net_plugin.cpp b/plugins/net_plugin/net_plugin.cpp
index 1e0e027d5e..5e8909cb91 100644
--- a/plugins/net_plugin/net_plugin.cpp
+++ b/plugins/net_plugin/net_plugin.cpp
@@ -785,7 +785,7 @@ namespace eosio {
       void handle_message( const packed_transaction& msg ) = delete; // packed_transaction_ptr overload used instead
       void handle_message( packed_transaction_ptr msg );
 
-      void process_signed_block( const block_id_type& id, signed_block_ptr msg );
+      void process_signed_block( const block_id_type& id, signed_block_ptr msg, block_state_ptr bsp );
 
       fc::variant_object get_logger_variant() const {
          fc::mutable_variant_object mvo;
@@ -3135,13 +3135,44 @@ namespace eosio {
    // called from connection strand
    void connection::handle_message( const block_id_type& id, signed_block_ptr ptr ) {
       peer_dlog( this, "received signed_block ${num}, id ${id}", ("num", ptr->block_num())("id", id) );
-      app().post(priority::medium, [ptr{std::move(ptr)}, id, c = shared_from_this()]() mutable {
-         c->process_signed_block( id, std::move( ptr ) );
+
+      controller& cc = my_impl->chain_plug->chain();
+      block_state_ptr bsp;
+      bool exception = false;
+      try {
+         if( cc.fetch_block_state_by_id( id ) ) {
+            my_impl->dispatcher->add_peer_block( id, connection_id );
+            my_impl->sync_master->sync_recv_block( shared_from_this(), id, ptr->block_num(), false );
+            return;
+         }
+         // this may return null if block is not immediately ready to be processed
+         bsp = cc.create_block_state( id, ptr );
+      } catch( const fc::exception& ex ) {
+         exception = true;
+         peer_elog(this, "bad block exception: #${n} ${id}...: ${m}",
+                   ("n", ptr->block_num())("id", id.str().substr(8,16))("m",ex.to_string()));
+      } catch( ... ) {
+         exception = true;
+         peer_elog(this, "bad block: #${n} ${id}...: unknown exception",
+                   ("n", ptr->block_num())("id", id.str().substr(8,16)));
+      }
+      if( exception ) {
+         my_impl->sync_master->rejected_block( shared_from_this(), ptr->block_num() );
+         my_impl->dispatcher->rejected_block( id );
+         return;
+      }
+
+      bool signal_producer = !!bsp; // ready to process immediately, so signal producer to interrupt start_block
+      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();
    }
 
    // called from application thread
-   void connection::process_signed_block( const block_id_type& blk_id, signed_block_ptr msg ) {
+   void connection::process_signed_block( const block_id_type& blk_id, signed_block_ptr msg, block_state_ptr bsp ) {
       controller& cc = my_impl->chain_plug->chain();
       uint32_t blk_num = msg->block_num();
       // use c in this method instead of this to highlight that all methods called on c-> must be thread safe
@@ -3171,7 +3202,7 @@ namespace eosio {
 
       go_away_reason reason = fatal_other;
       try {
-         bool accepted = my_impl->chain_plug->accept_block(msg, blk_id);
+         bool accepted = my_impl->chain_plug->accept_block(msg, blk_id, bsp);
          my_impl->update_chain_info();
          if( !accepted ) return;
          reason = no_reason;
diff --git a/plugins/producer_plugin/include/eosio/producer_plugin/producer_plugin.hpp b/plugins/producer_plugin/include/eosio/producer_plugin/producer_plugin.hpp
index 2ee1f132f6..bf54bc514d 100644
--- a/plugins/producer_plugin/include/eosio/producer_plugin/producer_plugin.hpp
+++ b/plugins/producer_plugin/include/eosio/producer_plugin/producer_plugin.hpp
@@ -144,6 +144,9 @@ class producer_plugin : public appbase::plugin<producer_plugin> {
 
    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();
+
  private:
    std::shared_ptr<class producer_plugin_impl> my;
 };
diff --git a/plugins/producer_plugin/include/eosio/producer_plugin/subjective_billing.hpp b/plugins/producer_plugin/include/eosio/producer_plugin/subjective_billing.hpp
index ad72825f3f..5e794466d6 100644
--- a/plugins/producer_plugin/include/eosio/producer_plugin/subjective_billing.hpp
+++ b/plugins/producer_plugin/include/eosio/producer_plugin/subjective_billing.hpp
@@ -180,7 +180,8 @@ class subjective_billing {
       fc_dlog( log, "Subjective billed accounts ${n} removed ${r}", ("n", orig_count)("r", orig_count - _account_subjective_bill_cache.size()) );
    }
 
-   bool remove_expired( fc::logger& log, const fc::time_point& pending_block_time, const fc::time_point& now, const fc::time_point& deadline ) {
+   template <typename Yield>
+   bool remove_expired( fc::logger& log, const fc::time_point& pending_block_time, const fc::time_point& now, Yield&& yield ) {
       bool exhausted = false;
       auto& idx = _trx_cache_index.get<by_expiry>();
       if( !idx.empty() ) {
@@ -189,7 +190,7 @@ class subjective_billing {
          uint32_t num_expired = 0;
 
          while( !idx.empty() ) {
-            if( deadline <= fc::time_point::now() ) {
+            if( yield() ) {
                exhausted = true;
                break;
             }
diff --git a/plugins/producer_plugin/producer_plugin.cpp b/plugins/producer_plugin/producer_plugin.cpp
index ef7161331b..4f8777d5fc 100644
--- a/plugins/producer_plugin/producer_plugin.cpp
+++ b/plugins/producer_plugin/producer_plugin.cpp
@@ -310,6 +310,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
       fc::microseconds                                          _max_irreversible_block_age_us;
       int32_t                                                   _produce_time_offset_us = 0;
       int32_t                                                   _last_block_time_offset_us = 0;
@@ -327,9 +328,6 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
 
       chain_plugin* chain_plug = nullptr;
 
-      incoming::channels::block::channel_type::handle         _incoming_block_subscription;
-      incoming::channels::transaction::channel_type::handle   _incoming_transaction_subscription;
-
       compat::channels::transaction_ack::channel_type&        _transaction_ack_channel;
 
       incoming::methods::block_sync::method_type::handle        _incoming_block_sync_provider;
@@ -424,7 +422,7 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
          _idle_trx_time = fc::time_point::now();
       }
 
-      bool on_incoming_block(const signed_block_ptr& block, const std::optional<block_id_type>& block_id) {
+      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 ) {
             fc_wlog( _log, "dropped incoming block #${num} id: ${id}",
@@ -432,6 +430,11 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
             return false;
          }
 
+         // start a new speculative block, speculative start_block may have been interrupted
+         auto ensure = fc::make_scoped_exit([this](){
+            schedule_production_loop();
+         });
+
          const auto& id = block_id ? *block_id : block->calculate_id();
          auto blk_num = block->block_num();
 
@@ -445,16 +448,14 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
          if( existing ) { return false; }
 
          // start processing of block
-         auto bsf = chain.create_block_state_future( id, block );
+         std::future<block_state_ptr> bsf;
+         if( !bsp ) {
+            bsf = chain.create_block_state_future( id, block );
+         }
 
          // abort the pending block
          abort_block();
 
-         // exceptions throw out, make sure we restart our loop
-         auto ensure = fc::make_scoped_exit([this](){
-            schedule_production_loop();
-         });
-
          // push the new block
          auto handle_error = [&](const auto& e)
          {
@@ -465,7 +466,8 @@ class producer_plugin_impl : public std::enable_shared_from_this<producer_plugin
 
          controller::block_report br;
          try {
-            chain.push_block( br, bsf, [this]( const branch_type& forked_branch ) {
+            const block_state_ptr& bspr = bsp ? bsp : bsf.get();
+            chain.push_block( br, bspr, [this]( const branch_type& forked_branch ) {
                _unapplied_transactions.add_forked( forked_branch );
             }, [this]( const transaction_id_type& id ) {
                return _unapplied_transactions.get_trx( id );
@@ -668,6 +670,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;
       start_block_result start_block();
 
       fc::time_point calculate_pending_block_time() const;
@@ -957,23 +960,9 @@ void producer_plugin::plugin_initialize(const boost::program_options::variables_
       }
    }
 
-   my->_incoming_block_subscription = app().get_channel<incoming::channels::block>().subscribe(
-         [this](const signed_block_ptr& block) {
-      try {
-         my->on_incoming_block(block, {});
-      } LOG_AND_DROP();
-   });
-
-   my->_incoming_transaction_subscription = app().get_channel<incoming::channels::transaction>().subscribe(
-         [this](const packed_transaction_ptr& trx) {
-      try {
-         my->on_incoming_transaction_async(trx, false, false, false, [](const auto&){});
-      } LOG_AND_DROP();
-   });
-
    my->_incoming_block_sync_provider = app().get_method<incoming::methods::block_sync>().register_provider(
-         [this](const signed_block_ptr& block, const std::optional<block_id_type>& block_id) {
-      return my->on_incoming_block(block, block_id);
+         [this](const signed_block_ptr& block, const std::optional<block_id_type>& block_id, const block_state_ptr& bsp) {
+      return my->on_incoming_block(block, block_id, bsp);
    });
 
    my->_incoming_transaction_async_provider = app().get_method<incoming::methods::transaction_async>().register_provider(
@@ -1588,6 +1577,14 @@ 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 ) {
+      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;
+}
+
 producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
    chain::controller& chain = chain_plug->chain();
 
@@ -1763,12 +1760,16 @@ producer_plugin_impl::start_block_result producer_plugin_impl::start_block() {
       }
 
       try {
+         _account_fails.clear();
+
          if( !remove_expired_trxs( preprocess_deadline ) )
             return start_block_result::exhausted;
          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(), preprocess_deadline ) )
+         if( !_subjective_billing.remove_expired( _log, chain.pending_block_time(), fc::time_point::now(),
+                                                  [&](){ return should_interrupt_start_block( preprocess_deadline ); } ) ) {
             return start_block_result::exhausted;
+         }
 
          // limit execution of pending incoming to once per block
          auto incoming_itr = _unapplied_transactions.incoming_begin();
@@ -1790,7 +1791,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 (preprocess_deadline <= fc::time_point::now() || block_is_exhausted()) {
+         if ( should_interrupt_start_block( preprocess_deadline ) || block_is_exhausted() ) {
             return start_block_result::exhausted;
          }
 
@@ -1822,7 +1823,7 @@ bool producer_plugin_impl::remove_expired_trxs( const fc::time_point& deadline )
    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, deadline,
+   bool exhausted = !_unapplied_transactions.clear_expired( pending_block_time, [&](){ return should_interrupt_start_block(deadline); },
          [&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 ) {
@@ -1832,7 +1833,7 @@ bool producer_plugin_impl::remove_expired_trxs( const fc::time_point& deadline )
             }
    });
 
-   if( exhausted ) {
+   if( exhausted && _pending_block_mode == pending_block_mode::producing ) {
       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) );
@@ -1858,7 +1859,7 @@ bool producer_plugin_impl::remove_expired_blacklisted_trxs( const fc::time_point
       int orig_count = _blacklisted_transactions.size();
 
       while (!blacklist_by_expiry.empty() && blacklist_by_expiry.begin()->expiry <= lib_time) {
-         if (deadline <= fc::time_point::now()) {
+         if ( should_interrupt_start_block( deadline ) ) {
             exhausted = true;
             break;
          }
@@ -2099,7 +2100,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( deadline <= fc::time_point::now() ) {
+         if( should_interrupt_start_block( deadline ) ) {
             exhausted = true;
             break;
          }
@@ -2281,7 +2282,7 @@ bool producer_plugin_impl::process_incoming_trxs( const fc::time_point& deadline
       size_t processed = 0;
       fc_dlog( _log, "Processing ${n} pending transactions", ("n", _unapplied_transactions.incoming_size()) );
       while( itr != end ) {
-         if (deadline <= fc::time_point::now()) {
+         if ( should_interrupt_start_block( deadline ) ) {
             exhausted = true;
             break;
          }
@@ -2327,6 +2328,7 @@ 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();
@@ -2532,6 +2534,10 @@ void producer_plugin_impl::produce_block() {
         ("confs", new_bs->header.confirmed));
 }
 
+void producer_plugin::received_block() {
+   my->_received_block = true;
+}
+
 void producer_plugin::log_failed_transaction(const transaction_id_type& trx_id, const packed_transaction_ptr& packed_trx_ptr, const char* reason) const {
    fc_dlog(_trx_log, "[TRX_TRACE] Speculative execution is REJECTING tx: ${trx}",
            ("entire_trx", packed_trx_ptr ? my->chain_plug->get_log_trx(packed_trx_ptr->get_transaction()) : fc::variant{trx_id}));
diff --git a/plugins/producer_plugin/test/test_subjective_billing.cpp b/plugins/producer_plugin/test/test_subjective_billing.cpp
index e8cffa9579..82273b1340 100644
--- a/plugins/producer_plugin/test/test_subjective_billing.cpp
+++ b/plugins/producer_plugin/test/test_subjective_billing.cpp
@@ -50,7 +50,7 @@ BOOST_AUTO_TEST_CASE( subjective_bill_test ) {
       BOOST_CHECK_EQUAL( 9, sub_bill.get_subjective_bill(b, now) );
 
       // expires transactions but leaves them in the decay at full value
-      sub_bill.remove_expired( log, now + fc::microseconds(1), now, fc::time_point::maximum() );
+      sub_bill.remove_expired( log, now + fc::microseconds(1), now, [](){ return false; } );
 
       BOOST_CHECK_EQUAL( 13+11, sub_bill.get_subjective_bill(a, now) );
       BOOST_CHECK_EQUAL( 9, sub_bill.get_subjective_bill(b, now) );
@@ -138,7 +138,7 @@ BOOST_AUTO_TEST_CASE( subjective_bill_test ) {
       BOOST_CHECK_EQUAL( 1024 + 1024, sub_bill.get_subjective_bill(a, now) );
       BOOST_CHECK_EQUAL( 1024, sub_bill.get_subjective_bill(b, now) );
 
-      sub_bill.remove_expired( log, now, now, fc::time_point::maximum() );
+      sub_bill.remove_expired( log, now, now, [](){ return false; } );
       BOOST_CHECK_EQUAL( 1024 + 1024, sub_bill.get_subjective_bill(a, now) );
       BOOST_CHECK_EQUAL( 1024, sub_bill.get_subjective_bill(b, now) );
 
@@ -148,7 +148,7 @@ BOOST_AUTO_TEST_CASE( subjective_bill_test ) {
       BOOST_CHECK_EQUAL( 1024, sub_bill.get_subjective_bill(a, endtime) );
       BOOST_CHECK_EQUAL( 0, sub_bill.get_subjective_bill(b, endtime) );
 
-      sub_bill.remove_expired( log, now + fc::microseconds(1), now, fc::time_point::maximum() );
+      sub_bill.remove_expired( log, now + fc::microseconds(1), now, [](){ return false; } );
       BOOST_CHECK_EQUAL( 1024 + 1024, sub_bill.get_subjective_bill(a, now) );
       BOOST_CHECK_EQUAL( 1024, sub_bill.get_subjective_bill(b, now) );
 
diff --git a/unittests/block_tests.cpp b/unittests/block_tests.cpp
index f441fc95af..2bf0b503af 100644
--- a/unittests/block_tests.cpp
+++ b/unittests/block_tests.cpp
@@ -44,10 +44,10 @@ BOOST_AUTO_TEST_CASE(block_with_invalid_tx_test)
 
    // Push block with invalid transaction to other chain
    tester validator;
-   auto bs = validator.control->create_block_state_future( copy_b->calculate_id(), copy_b );
+   auto bsf = validator.control->create_block_state_future( copy_b->calculate_id(), copy_b );
    validator.control->abort_block();
    controller::block_report br;
-   BOOST_REQUIRE_EXCEPTION(validator.control->push_block( br, bs, forked_branch_callback{}, trx_meta_cache_lookup{} ), fc::exception ,
+   BOOST_REQUIRE_EXCEPTION(validator.control->push_block( br, bsf.get(), forked_branch_callback{}, trx_meta_cache_lookup{} ), fc::exception ,
    [] (const fc::exception &e)->bool {
       return e.code() == account_name_exists_exception::code_value ;
    }) ;
@@ -83,10 +83,10 @@ BOOST_AUTO_TEST_CASE(block_with_invalid_tx_mroot_test)
 
    // Push block with invalid transaction to other chain
    tester validator;
-   auto bs = validator.control->create_block_state_future( copy_b->calculate_id(), copy_b );
+   auto bsf = validator.control->create_block_state_future( copy_b->calculate_id(), copy_b );
    validator.control->abort_block();
    controller::block_report br;
-   BOOST_REQUIRE_EXCEPTION(validator.control->push_block( br, bs, forked_branch_callback{}, trx_meta_cache_lookup{} ), fc::exception ,
+   BOOST_REQUIRE_EXCEPTION(validator.control->push_block( br, bsf.get(), forked_branch_callback{}, trx_meta_cache_lookup{} ), fc::exception ,
                            [] (const fc::exception &e)->bool {
                               return e.code() == block_validate_exception::code_value &&
                                      e.to_detail_string().find("invalid block transaction merkle root") != std::string::npos;
diff --git a/unittests/forked_tests.cpp b/unittests/forked_tests.cpp
index ebb321fc71..349ed67137 100644
--- a/unittests/forked_tests.cpp
+++ b/unittests/forked_tests.cpp
@@ -267,10 +267,10 @@ BOOST_AUTO_TEST_CASE( forking ) try {
    signed_block bad_block = std::move(*b);
    bad_block.action_mroot = bad_block.previous;
    auto bad_id = bad_block.calculate_id();
-   auto bad_block_bs = c.control->create_block_state_future( bad_id, std::make_shared<signed_block>(std::move(bad_block)) );
+   auto bad_block_bsf = c.control->create_block_state_future( bad_id, std::make_shared<signed_block>(std::move(bad_block)) );
    c.control->abort_block();
    controller::block_report br;
-   BOOST_REQUIRE_EXCEPTION(c.control->push_block( br, bad_block_bs, forked_branch_callback{}, trx_meta_cache_lookup{} ), fc::exception,
+   BOOST_REQUIRE_EXCEPTION(c.control->push_block( br, bad_block_bsf.get(), forked_branch_callback{}, trx_meta_cache_lookup{} ), fc::exception,
       [] (const fc::exception &ex)->bool {
          return ex.to_detail_string().find("block signed by unexpected key") != std::string::npos;
       });
diff --git a/unittests/unapplied_transaction_queue_tests.cpp b/unittests/unapplied_transaction_queue_tests.cpp
index a18d9c50e4..21614c56bb 100644
--- a/unittests/unapplied_transaction_queue_tests.cpp
+++ b/unittests/unapplied_transaction_queue_tests.cpp
@@ -339,7 +339,7 @@ BOOST_AUTO_TEST_CASE( unapplied_transaction_queue_test ) try {
    q.add_aborted( { trx20, trx22 } );
    q.add_persisted( trx21 );
    q.add_persisted( trx23 );
-   q.clear_expired( fc::time_point::now(), fc::time_point::now() + fc::seconds( 300 ), [](auto, auto){} );
+   q.clear_expired( fc::time_point::now(), [](){ return false; }, [](auto, auto){} );
    BOOST_CHECK( q.size() == 2 );
    BOOST_REQUIRE( next( q ) == trx23 );
    BOOST_REQUIRE( next( q ) == trx22 );