Remove block_arrival

nano/core_test/node.cpp

@@ -2136,53 +2136,6 @@ TEST (node, block_confirm)
 	ASSERT_TIMELY (10s, node1.active.recently_cemented.list ().size () == 1);
 }
 
-TEST (node, block_arrival)
-{
-	nano::test::system system (1);
-	auto & node (*system.nodes[0]);
-	ASSERT_EQ (0, node.block_arrival.arrival.size ());
-	nano::block_hash hash1 (1);
-	node.block_arrival.add (hash1);
-	ASSERT_EQ (1, node.block_arrival.arrival.size ());
-	node.block_arrival.add (hash1);
-	ASSERT_EQ (1, node.block_arrival.arrival.size ());
-	nano::block_hash hash2 (2);
-	node.block_arrival.add (hash2);
-	ASSERT_EQ (2, node.block_arrival.arrival.size ());
-}
-
-TEST (node, block_arrival_size)
-{
-	nano::test::system system (1);
-	auto & node (*system.nodes[0]);
-	auto time (std::chrono::steady_clock::now () - nano::block_arrival::arrival_time_min - std::chrono::seconds (5));
-	nano::block_hash hash (0);
-	for (auto i (0); i < nano::block_arrival::arrival_size_min * 2; ++i)
-	{
-		node.block_arrival.arrival.push_back (nano::block_arrival_info{ time, hash });
-		++hash.qwords[0];
-	}
-	ASSERT_EQ (nano::block_arrival::arrival_size_min * 2, node.block_arrival.arrival.size ());
-	node.block_arrival.recent (0);
-	ASSERT_EQ (nano::block_arrival::arrival_size_min, node.block_arrival.arrival.size ());
-}
-
-TEST (node, block_arrival_time)
-{
-	nano::test::system system (1);
-	auto & node (*system.nodes[0]);
-	auto time (std::chrono::steady_clock::now ());
-	nano::block_hash hash (0);
-	for (auto i (0); i < nano::block_arrival::arrival_size_min * 2; ++i)
-	{
-		node.block_arrival.arrival.push_back (nano::block_arrival_info{ time, hash });
-		++hash.qwords[0];
-	}
-	ASSERT_EQ (nano::block_arrival::arrival_size_min * 2, node.block_arrival.arrival.size ());
-	node.block_arrival.recent (0);
-	ASSERT_EQ (nano::block_arrival::arrival_size_min * 2, node.block_arrival.arrival.size ());
-}
-
 TEST (node, confirm_quorum)
 {
 	nano::test::system system (1);

nano/lib/locks.cpp

@@ -250,8 +250,6 @@ char const * nano::mutex_identifier (mutexes mutex)
 	{
 		case mutexes::active:
 			return "active";
-		case mutexes::block_arrival:
-			return "block_arrival";
 		case mutexes::block_processor:
 			return "block_processor";
 		case mutexes::block_uniquer:

nano/lib/locks.hpp

@@ -20,7 +20,6 @@ bool any_filters_registered ();
 enum class mutexes
 {
 	active,
-	block_arrival,
 	block_processor,
 	block_uniquer,
 	blockstore_cache,

nano/node/CMakeLists.txt

@@ -20,8 +20,6 @@ add_library(
   backlog_population.cpp
   bandwidth_limiter.hpp
   bandwidth_limiter.cpp
-  block_arrival.hpp
-  block_arrival.cpp
   block_broadcast.cpp
   block_broadcast.hpp
   blocking_observer.cpp

nano/node/block_broadcast.cpp

@@ -1,14 +1,12 @@
 #include <nano/lib/threading.hpp>
 #include <nano/lib/utility.hpp>
-#include <nano/node/block_arrival.hpp>
 #include <nano/node/block_broadcast.hpp>
 #include <nano/node/blockprocessor.hpp>
 #include <nano/node/network.hpp>
 
-nano::block_broadcast::block_broadcast (nano::block_processor & block_processor_a, nano::network & network_a, nano::block_arrival & block_arrival_a, nano::stats & stats_a, bool enabled_a) :
+nano::block_broadcast::block_broadcast (nano::block_processor & block_processor_a, nano::network & network_a, nano::stats & stats_a, bool enabled_a) :
 	block_processor{ block_processor_a },
 	network{ network_a },
-	block_arrival{ block_arrival_a },
 	stats{ stats_a },
 	enabled{ enabled_a },
 	queue{ stats_a, nano::stat::type::block_broadcaster, nano::thread_role::name::block_broadcasting, /* single thread */ 1, max_size }
@@ -22,7 +20,7 @@ nano::block_broadcast::block_broadcast (nano::block_processor & block_processor_
 		switch (result.code)
 		{
 			case nano::process_result::progress:
-				observe (block);
+				observe (block, context);
 				break;
 			default:
 				break;
@@ -54,7 +52,7 @@ void nano::block_broadcast::stop ()
 	queue.stop ();
 }
 
-void nano::block_broadcast::observe (std::shared_ptr<nano::block> const & block)
+void nano::block_broadcast::observe (std::shared_ptr<nano::block> const & block, nano::block_processor::context const & context)
 {
 	bool is_local = local.contains (block->hash ());
 	if (is_local)
@@ -65,7 +63,7 @@ void nano::block_broadcast::observe (std::shared_ptr<nano::block> const & block)
 	}
 	else
 	{
-		if (block_arrival.recent (block->hash ()))
+		if (context.recent_arrival ())
 		{
 			// Block arrived from realtime traffic, do normal gossip.
 			queue.add (entry{ block, broadcast_strategy::normal });

nano/node/block_broadcast.hpp

@@ -3,9 +3,12 @@
 #include <nano/lib/blocks.hpp>
 #include <nano/lib/locks.hpp>
 #include <nano/lib/processing_queue.hpp>
+#include <nano/node/blockprocessor.hpp>
+#include <nano/secure/common.hpp>
 
 #include <boost/multi_index/hashed_index.hpp>
 #include <boost/multi_index/ordered_index.hpp>
+#include <boost/multi_index/sequenced_index.hpp>
 #include <boost/multi_index_container.hpp>
 
 #include <memory>
@@ -16,8 +19,6 @@ namespace mi = boost::multi_index;
 
 namespace nano
 {
-class block_arrival;
-class block_processor;
 class network;
 }
 
@@ -39,7 +40,7 @@ class block_broadcast
 	using queue_t = nano::processing_queue<entry>;
 
 public:
-	block_broadcast (nano::block_processor &, nano::network &, nano::block_arrival &, nano::stats &, bool enabled = false);
+	block_broadcast (nano::block_processor &, nano::network &, nano::stats &, bool enabled = false);
 	~block_broadcast ();
 
 	void start ();
@@ -51,12 +52,11 @@ class block_broadcast
 private: // Dependencies
 	nano::block_processor & block_processor;
 	nano::network & network;
-	nano::block_arrival & block_arrival;
 	nano::stats & stats;
 
 private:
 	// Block_processor observer
-	void observe (std::shared_ptr<nano::block> const & block);
+	void observe (std::shared_ptr<nano::block> const & block, nano::block_processor::context const & context);
 	void process_batch (queue_t::batch_t & batch);
 
 private:

nano/node/blockprocessor.cpp

@@ -6,6 +6,10 @@
 
 #include <boost/format.hpp>
 
+/*
+ * block_processor
+ */
+
 nano::block_processor::block_processor (nano::node & node_a, nano::write_database_queue & write_database_queue_a) :
 	next_log (std::chrono::steady_clock::now ()),
 	node (node_a),
@@ -443,3 +447,12 @@ std::unique_ptr<nano::container_info_component> nano::collect_container_info (bl
 	composite->add_component (std::make_unique<container_info_leaf> (container_info{ "forced", forced_count, sizeof (decltype (block_processor.forced)::value_type) }));
 	return composite;
 }
+
+/*
+ * context
+ */
+
+bool nano::block_processor::context::recent_arrival () const
+{
+	return std::chrono::steady_clock::now () < arrival + recent_arrival_cutoff;
+}

nano/node/blockprocessor.hpp

@@ -39,6 +39,10 @@ class block_processor final
 	{
 		block_source source;
 		std::chrono::steady_clock::time_point arrival;
+
+	public:
+		bool recent_arrival () const;
+		static std::chrono::seconds constexpr recent_arrival_cutoff{ 60 * 5 };
 	};
 
 	using entry_t = std::pair<std::shared_ptr<nano::block>, context>;

nano/node/json_handler.cpp

@@ -1214,7 +1214,6 @@ void nano::json_handler::block_confirm ()
 				nano::election_status status{ block_l, 0, 0, std::chrono::duration_cast<std::chrono::milliseconds> (std::chrono::system_clock::now ().time_since_epoch ()), std::chrono::duration_values<std::chrono::milliseconds>::zero (), 0, 1, 0, nano::election_status_type::active_confirmation_height };
 				node.active.recently_cemented.put (status);
 				// Trigger callback for confirmed block
-				node.block_arrival.add (hash);
 				auto account (node.ledger.account (transaction, hash));
 				bool error_or_pruned (false);
 				auto amount (node.ledger.amount_safe (transaction, hash, error_or_pruned));

nano/node/node.cpp

@@ -195,7 +195,7 @@ nano::node::node (boost::asio::io_context & io_ctx_a, std::filesystem::path cons
 	epoch_upgrader{ *this, ledger, store, network_params, logger },
 	startup_time (std::chrono::steady_clock::now ()),
 	node_seq (seq),
-	block_broadcast{ block_processor, network, block_arrival, stats, !flags.disable_block_processor_republishing },
+	block_broadcast{ block_processor, network, stats, !flags.disable_block_processor_republishing },
 	process_live_dispatcher{ ledger, scheduler.priority, vote_cache, websocket }
 {
 	process_live_dispatcher.connect (block_processor);
@@ -236,7 +236,7 @@ nano::node::node (boost::asio::io_context & io_ctx_a, std::filesystem::path cons
 		{
 			observers.blocks.add ([this] (nano::election_status const & status_a, std::vector<nano::vote_with_weight_info> const & votes_a, nano::account const & account_a, nano::amount const & amount_a, bool is_state_send_a, bool is_state_epoch_a) {
 				auto block_a (status_a.winner);
-				if ((status_a.type == nano::election_status_type::active_confirmed_quorum || status_a.type == nano::election_status_type::active_confirmation_height) && this->block_arrival.recent (block_a->hash ()))
+				if ((status_a.type == nano::election_status_type::active_confirmed_quorum || status_a.type == nano::election_status_type::active_confirmation_height))
 				{
 					auto node_l (shared_from_this ());
 					background ([node_l, block_a, account_a, amount_a, is_state_send_a, is_state_epoch_a] () {
@@ -555,7 +555,6 @@ std::unique_ptr<nano::container_info_component> nano::collect_container_info (no
 	composite->add_component (collect_container_info (node.vote_processor, "vote_processor"));
 	composite->add_component (collect_container_info (node.rep_crawler, "rep_crawler"));
 	composite->add_component (collect_container_info (node.block_processor, "block_processor"));
-	composite->add_component (collect_container_info (node.block_arrival, "block_arrival"));
 	composite->add_component (collect_container_info (node.online_reps, "online_reps"));
 	composite->add_component (collect_container_info (node.history, "history"));
 	composite->add_component (node.block_uniquer.collect_container_info ("block_uniquer"));
@@ -574,7 +573,6 @@ std::unique_ptr<nano::container_info_component> nano::collect_container_info (no
 
 void nano::node::process_active (std::shared_ptr<nano::block> const & incoming)
 {
-	block_arrival.add (incoming->hash ());
 	block_processor.add (incoming);
 }
 
@@ -591,18 +589,13 @@ nano::process_return nano::node::process (nano::block & block)
 
 std::optional<nano::process_return> nano::node::process_local (std::shared_ptr<nano::block> const & block_a)
 {
-	// Add block hash as recently arrived to trigger automatic rebroadcast and election
-	block_arrival.add (block_a->hash ());
 	block_broadcast.track_local (block_a->hash ());
 	return block_processor.add_blocking (block_a);
 }
 
 void nano::node::process_local_async (std::shared_ptr<nano::block> const & block_a)
 {
-	// Add block hash as recently arrived to trigger automatic rebroadcast and election
-	block_arrival.add (block_a->hash ());
-	// Set current time to trigger automatic rebroadcast and election
-	block_processor.add (block_a);
+	block_processor.add (block_a, nano::block_processor::block_source::local);
 }
 
 void nano::node::start ()

nano/node/node.hpp

@@ -7,7 +7,6 @@
 #include <nano/node/active_transactions.hpp>
 #include <nano/node/backlog_population.hpp>
 #include <nano/node/bandwidth_limiter.hpp>
-#include <nano/node/block_arrival.hpp>
 #include <nano/node/block_broadcast.hpp>
 #include <nano/node/blockprocessor.hpp>
 #include <nano/node/bootstrap/bootstrap.hpp>
@@ -168,7 +167,6 @@ class node final : public std::enable_shared_from_this<nano::node>
 	nano::vote_processor vote_processor;
 	unsigned warmed_up;
 	nano::block_processor block_processor;
-	nano::block_arrival block_arrival;
 	nano::local_vote_history history;
 	nano::keypair node_id;
 	nano::block_uniquer block_uniquer;

nano/rpc_test/rpc.cpp

@@ -4429,7 +4429,6 @@ TEST (rpc, populate_backlog)
 				.work (*node->work_generate_blocking (latest))
 				.build ();
 	ASSERT_EQ (nano::process_result::progress, node->process (*send).code);
-	ASSERT_FALSE (node->block_arrival.recent (send->hash ()));
 
 	auto const rpc_ctx = add_rpc (system, node);
 	boost::property_tree::ptree request;