Block arrival activate (#3970)

This patch fixes an issue where an election might not be started on time during non-bootstrapping situations.

nano/core_test/active_transactions.cpp

@@ -457,11 +457,10 @@ TEST (active_transactions, inactive_votes_cache_election_start)
 				 .sign (key2.prv, key2.pub)
 				 .work (*system.work.generate (key2.pub))
 				 .build_shared ();
-	node.block_processor.add (send1);
+	ASSERT_EQ (nano::process_result::progress, node.process (*send1).code);
-	node.block_processor.add (send2);
+	ASSERT_EQ (nano::process_result::progress, node.process (*send2).code);
-	node.block_processor.add (open1);
+	ASSERT_EQ (nano::process_result::progress, node.process (*open1).code);
-	node.block_processor.add (open2);
+	ASSERT_EQ (nano::process_result::progress, node.process (*open2).code);
-	node.block_processor.flush ();
 	ASSERT_TIMELY (5s, 5 == node.ledger.cache.block_count);
 	ASSERT_TRUE (node.active.empty ());
 	ASSERT_EQ (1, node.ledger.cache.cemented_count);
@@ -1651,4 +1650,4 @@ TEST (active_transactions, allow_limited_overflow_adapt)
 	ASSERT_TIMELY_EQ (5s, node.active.size (), node.active.limit ());
 	// And it stays that way without increasing
 	ASSERT_ALWAYS (1s, node.active.size () == node.active.limit ());
-}
+}

nano/core_test/bootstrap.cpp

@@ -317,7 +317,6 @@ TEST (bootstrap_processor, process_one)
 	node1->bootstrap_initiator.bootstrap (node0->network.endpoint (), false);
 	ASSERT_NE (node1->latest (nano::dev::genesis_key.pub), node0->latest (nano::dev::genesis_key.pub));
 	ASSERT_TIMELY (10s, node1->latest (nano::dev::genesis_key.pub) == node0->latest (nano::dev::genesis_key.pub));
-	ASSERT_EQ (0, node1->active.size ());
 	node1->stop ();
 }
 
@@ -384,13 +383,13 @@ TEST (bootstrap_processor, process_state)
 	ASSERT_EQ (nano::process_result::progress, node0->process (*block1).code);
 	ASSERT_EQ (nano::process_result::progress, node0->process (*block2).code);
 
+	config.peering_port = nano::test::get_available_port ();
 	auto node1 (std::make_shared<nano::node> (system.io_ctx, nano::test::get_available_port (), nano::unique_path (), system.logging, system.work, node_flags));
 	ASSERT_EQ (node0->latest (nano::dev::genesis_key.pub), block2->hash ());
 	ASSERT_NE (node1->latest (nano::dev::genesis_key.pub), block2->hash ());
 	node1->bootstrap_initiator.bootstrap (node0->network.endpoint (), false);
 	ASSERT_NE (node1->latest (nano::dev::genesis_key.pub), node0->latest (nano::dev::genesis_key.pub));
 	ASSERT_TIMELY (10s, node1->latest (nano::dev::genesis_key.pub) == node0->latest (nano::dev::genesis_key.pub));
-	ASSERT_TIMELY (10s, node1->active.empty ());
 	node1->stop ();
 }
 

nano/core_test/ledger.cpp

@@ -4377,7 +4377,6 @@ TEST (ledger, unchecked_epoch_invalid)
 	{
 		auto transaction = node1.store.tx_begin_read ();
 		ASSERT_FALSE (node1.store.block.exists (transaction, epoch1->hash ()));
-		ASSERT_TRUE (node1.active.empty ());
 		auto unchecked_count = node1.unchecked.count (transaction);
 		ASSERT_EQ (unchecked_count, 0);
 		ASSERT_EQ (unchecked_count, node1.unchecked.count (transaction));

nano/node/blockprocessor.cpp

@@ -329,7 +329,7 @@ void nano::block_processor::process_live (nano::transaction const & transaction_
 	{
 		node.network.flood_block_initial (block_a);
 	}
-	else if (!node.flags.disable_block_processor_republishing)
+	else if (!node.flags.disable_block_processor_republishing && node.block_arrival.recent (hash_a))
 	{
 		node.network.flood_block (block_a, nano::buffer_drop_policy::limiter);
 	}
@@ -360,10 +360,9 @@ nano::process_return nano::block_processor::process_one (nano::write_transaction
 				block->serialize_json (block_string, node.config.logging.single_line_record ());
 				node.logger.try_log (boost::str (boost::format ("Processing block %1%: %2%") % hash.to_string () % block_string));
 			}
-			if (node.block_arrival.recent (hash) || forced_a)
+			events_a.events.emplace_back ([this, hash, block = info_a.block, result, origin_a] (nano::transaction const & post_event_transaction_a) {
-			{
+				process_live (post_event_transaction_a, hash, block, result, origin_a);
-				events_a.events.emplace_back ([this, hash, block = info_a.block, result, origin_a] (nano::transaction const & post_event_transaction_a) { process_live (post_event_transaction_a, hash, block, result, origin_a); });
+			});
-			}
 			queue_unchecked (transaction_a, hash);
 			/* For send blocks check epoch open unchecked (gap pending).
 			For state blocks check only send subtype and only if block epoch is not last epoch.

nano/slow_test/node.cpp

@@ -2124,3 +2124,107 @@ TEST (node, wallet_create_block_confirm_conflicts)
 		t.join ();
 	}
 }
+
+/**
+ * This test creates a small network of evenly weighted PRs and ensures a sequence of blocks from the genesis account to random accounts are able to be processed
+ * Ongoing bootstrap is disabled to directly test election activation. A failure to activate a block on any PR will cause the test to stall
+ */
+TEST (system, block_sequence)
+{
+	size_t const block_count = 400;
+	size_t const pr_count = 4;
+	size_t const listeners_per_pr = 0;
+	nano::test::system system;
+	std::vector<nano::keypair> reps;
+	for (auto i = 0; i < pr_count; ++i)
+	{
+		reps.push_back (nano::keypair{});
+	}
+	system.ledger_initialization_set (reps, nano::Gxrb_ratio);
+	system.deadline_set (3600s);
+	nano::node_config config;
+	config.peering_port = nano::test::get_available_port ();
+	//config.bandwidth_limit = 16 * 1024;
+	config.enable_voting = true;
+	config.frontiers_confirmation = nano::frontiers_confirmation_mode::disabled;
+	nano::node_flags flags;
+	flags.disable_max_peers_per_ip = true;
+	flags.disable_ongoing_bootstrap = true;
+	auto root = system.add_node (config, flags);
+	config.preconfigured_peers.push_back ("::ffff:127.0.0.1:" + std::to_string (root->network.endpoint ().port ()));
+	auto wallet = root->wallets.items.begin ()->second;
+	wallet->insert_adhoc (nano::dev::genesis_key.prv);
+	for (auto rep : reps)
+	{
+		system.wallet (0);
+		config.peering_port = nano::test::get_available_port ();
+		auto pr = system.add_node (config, flags, nano::transport::transport_type::tcp, rep);
+		for (auto j = 0; j < listeners_per_pr; ++j)
+		{
+			config.peering_port = nano::test::get_available_port ();
+			system.add_node (config, flags);
+		}
+		std::cerr << rep.pub.to_account () << ' ' << pr->wallets.items.begin ()->second->exists (rep.pub) << pr->weight (rep.pub) << ' ' << '\n';
+	}
+	while (std::any_of (system.nodes.begin (), system.nodes.end (), [] (std::shared_ptr<nano::node> const & node) {
+		//std::cerr << node->rep_crawler.representative_count () << ' ';
+		return node->rep_crawler.representative_count () < 3;
+	}))
+	{
+		system.poll ();
+	}
+	for (auto & node : system.nodes)
+	{
+		std::cerr << std::to_string (node->network.port) << ": ";
+		auto prs = node->rep_crawler.principal_representatives ();
+		for (auto pr : prs)
+		{
+			std::cerr << pr.account.to_account () << ' ';
+		}
+		std::cerr << '\n';
+	}
+	nano::keypair key;
+	auto start = std::chrono::system_clock::now ();
+	std::deque<std::shared_ptr<nano::block>> blocks;
+	for (auto i = 0; i < block_count; ++i)
+	{
+		if ((i % 1000) == 0)
+		{
+			std::cerr << "Block: " << std::to_string (i) << " ms: " << std::to_string (std::chrono::duration_cast<std::chrono::milliseconds> (std::chrono::system_clock::now () - start).count ()) << "\n";
+		}
+		auto block = wallet->send_action (nano::dev::genesis_key.pub, key.pub, 1);
+		debug_assert (block != nullptr);
+		blocks.push_back (block);
+	}
+	auto done = false;
+	std::chrono::system_clock::time_point last;
+	auto interval = 1000ms;
+	while (!done)
+	{
+		if (std::chrono::system_clock::now () - last > interval)
+		{
+			std::string message;
+			for (auto i : system.nodes)
+			{
+				message += boost::str (boost::format ("N:%1% b:%2% c:%3% a:%4% s:%5% p:%6%\n") % std::to_string (i->network.port) % std::to_string (i->ledger.cache.block_count) % std::to_string (i->ledger.cache.cemented_count) % std::to_string (i->active.size ()) % std::to_string (i->scheduler.size ()) % std::to_string (i->network.size ()));
+				nano::lock_guard<nano::mutex> lock{ i->active.mutex };
+				for (auto const & j : i->active.roots)
+				{
+					auto election = j.election;
+					if (election->confirmation_request_count > 10)
+					{
+						message += boost::str (boost::format ("\t r:%1% i:%2%\n") % j.root.to_string () % std::to_string (election->confirmation_request_count));
+						for (auto const & k : election->votes ())
+						{
+							message += boost::str (boost::format ("\t\t r:%1% t:%2%\n") % k.first.to_account () % std::to_string (k.second.timestamp));
+						}
+					}
+				}
+			}
+			std::cerr << message << std::endl;
+			last = std::chrono::system_clock::now ();
+		}
+		done = std::all_of (system.nodes.begin (), system.nodes.end (), [&blocks] (std::shared_ptr<nano::node> node) { return node->block_confirmed (blocks.back ()->hash ()); });
+		system.poll ();
+	}
+}

nano/test_common/system.cpp

@@ -35,7 +35,7 @@ std::shared_ptr<nano::node> nano::test::system::add_node (nano::node_flags node_
 }
 
 /** Returns the node added. */
-std::shared_ptr<nano::node> nano::test::system::add_node (nano::node_config const & node_config_a, nano::node_flags node_flags_a, nano::transport::transport_type type_a)
+std::shared_ptr<nano::node> nano::test::system::add_node (nano::node_config const & node_config_a, nano::node_flags node_flags_a, nano::transport::transport_type type_a, std::optional<nano::keypair> const & rep)
 {
 	auto node (std::make_shared<nano::node> (io_ctx, nano::unique_path (), node_config_a, work, node_flags_a, node_sequence++));
 	for (auto i : initialization_blocks)
@@ -44,8 +44,12 @@ std::shared_ptr<nano::node> nano::test::system::add_node (nano::node_config cons
 		debug_assert (result.code == nano::process_result::progress);
 	}
 	debug_assert (!node->init_error ());
+	auto wallet = node->wallets.create (nano::random_wallet_id ());
+	if (rep)
+	{
+		wallet->insert_adhoc (rep->prv);
+	}
 	node->start ();
-	node->wallets.create (nano::random_wallet_id ());
 	nodes.reserve (nodes.size () + 1);
 	nodes.push_back (node);
 	if (nodes.size () > 1)

nano/test_common/system.hpp

@@ -56,7 +56,7 @@ namespace test
 		 */
 		nano::node & node (std::size_t index) const;
 		std::shared_ptr<nano::node> add_node (nano::node_flags = nano::node_flags (), nano::transport::transport_type = nano::transport::transport_type::tcp);
-		std::shared_ptr<nano::node> add_node (nano::node_config const &, nano::node_flags = nano::node_flags (), nano::transport::transport_type = nano::transport::transport_type::tcp);
+		std::shared_ptr<nano::node> add_node (nano::node_config const &, nano::node_flags = nano::node_flags (), nano::transport::transport_type = nano::transport::transport_type::tcp, std::optional<nano::keypair> const & rep = std::nullopt);
 		/*
 		 * Returns default config for node running in test environment
 		 */