Merge pull request #4854 from pwojcikdev/online-reps-work

Online weight tracking improvements

nano/core_test/election.cpp

@@ -198,8 +198,6 @@ TEST (election, quorum_minimum_confirm_fail)
 	ASSERT_FALSE (election->confirmed ());
 }
 
-namespace nano
-{
 // FIXME: this test fails on rare occasions. It needs a review.
 TEST (election, quorum_minimum_update_weight_before_quorum_checks)
 {
@@ -267,7 +265,6 @@ TEST (election, quorum_minimum_update_weight_before_quorum_checks)
 	ASSERT_TIMELY (5s, election->confirmed ());
 	ASSERT_NE (nullptr, node1.block (send1->hash ()));
 }
-}
 
 TEST (election, continuous_voting)
 {

nano/core_test/node.cpp

@@ -2783,9 +2783,8 @@ TEST (node, rollback_vote_self)
 		system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
 
 		// Without the rollback being finished, the aggregator should not reply with any vote
-		auto channel = std::make_shared<nano::transport::fake::channel> (node);
-		node.aggregator.request ({ { send2->hash (), send2->root () } }, channel);
-		ASSERT_ALWAYS_EQ (1s, node.stats.count (nano::stat::type::request_aggregator_replies), 0);
+		node.aggregator.request ({ { send2->hash (), send2->root () } }, node.loopback_channel);
+		ASSERT_ALWAYS (1s, !election->votes ().contains (nano::dev::genesis_key.pub));
 
 		// Going out of the scope allows the rollback to complete
 	}

nano/core_test/online_reps.cpp

@@ -1,3 +1,5 @@
+#include "nano/node/election.hpp"
+
 #include <nano/node/online_reps.hpp>
 #include <nano/node/transport/fake.hpp>
 #include <nano/secure/vote.hpp>
@@ -8,8 +10,8 @@
 
 TEST (online_reps, basic)
 {
-	nano::test::system system (1);
-	auto & node1 (*system.nodes[0]);
+	nano::test::system system;
+	auto & node1 = *system.add_node ();
 	// 1 sample of minimum weight
 	ASSERT_EQ (node1.config.online_weight_minimum, node1.online_reps.trended ());
 	auto vote (std::make_shared<nano::vote> ());
@@ -69,4 +71,204 @@ TEST (online_reps, election)
 	ASSERT_EQ (0, node1.online_reps.online ());
 	node1.vote_processor.vote_blocking (vote, std::make_shared<nano::transport::fake::channel> (node1));
 	ASSERT_EQ (nano::dev::constants.genesis_amount - nano::Knano_ratio, node1.online_reps.online ());
+}
+
+// Online reps should be able to observe remote representative
+TEST (online_reps, observe)
+{
+	nano::test::system system;
+	auto & node = *system.add_node ();
+	ASSERT_EQ (0, node.online_reps.online ());
+
+	// Addd genesis representative
+	auto & node_rep = *system.add_node ();
+	system.wallet (1)->insert_adhoc (nano::dev::genesis_key.prv);
+
+	// The node should see that weight as online
+	ASSERT_TIMELY_EQ (10s, node.online_reps.online (), nano::dev::constants.genesis_amount);
+	ASSERT_ALWAYS_EQ (1s, node.online_reps.online (), nano::dev::constants.genesis_amount);
+}
+
+TEST (online_reps, observe_multiple)
+{
+	nano::test::system system;
+	auto & node = *system.add_node ();
+	ASSERT_EQ (0, node.online_reps.online ());
+
+	auto & node_rep1 = *system.add_node (); // key1
+	auto & node_rep2 = *system.add_node (); // key2 & key3
+
+	auto const weight_1 = nano::nano_ratio * 1000;
+	auto const weight_2 = nano::nano_ratio * 1000000;
+	auto const weight_3 = nano::nano_ratio * 10000000;
+
+	nano::keypair key1, key2, key3;
+
+	// Distribute genesis voting weight
+	{
+		nano::block_builder builder;
+		auto send1 = builder.state ()
+					 .account (nano::dev::genesis_key.pub)
+					 .previous (nano::dev::genesis->hash ())
+					 .representative (nano::dev::genesis_key.pub)
+					 .balance (nano::dev::constants.genesis_amount - weight_1)
+					 .link (key1.pub)
+					 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
+					 .work (*system.work.generate (nano::dev::genesis->hash ()))
+					 .build ();
+		auto send2 = builder.state ()
+					 .account (nano::dev::genesis_key.pub)
+					 .previous (send1->hash ())
+					 .representative (nano::dev::genesis_key.pub)
+					 .balance (nano::dev::constants.genesis_amount - weight_1 - weight_2)
+					 .link (key2.pub)
+					 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
+					 .work (*system.work.generate (send1->hash ()))
+					 .build ();
+		auto send3 = builder.state ()
+					 .account (nano::dev::genesis_key.pub)
+					 .previous (send2->hash ())
+					 .representative (nano::dev::genesis_key.pub)
+					 .balance (nano::dev::constants.genesis_amount - weight_1 - weight_2 - weight_3)
+					 .link (key3.pub)
+					 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
+					 .work (*system.work.generate (send2->hash ()))
+					 .build ();
+		auto open1 = builder.state ()
+					 .account (key1.pub)
+					 .previous (0)
+					 .representative (key1.pub)
+					 .balance (weight_1)
+					 .link (send1->hash ())
+					 .sign (key1.prv, key1.pub)
+					 .work (*system.work.generate (key1.pub))
+					 .build ();
+		auto open2 = builder.state ()
+					 .account (key2.pub)
+					 .previous (0)
+					 .representative (key2.pub)
+					 .balance (weight_2)
+					 .link (send2->hash ())
+					 .sign (key2.prv, key2.pub)
+					 .work (*system.work.generate (key2.pub))
+					 .build ();
+		auto open3 = builder.state ()
+					 .account (key3.pub)
+					 .previous (0)
+					 .representative (key3.pub)
+					 .balance (weight_3)
+					 .link (send3->hash ())
+					 .sign (key3.prv, key3.pub)
+					 .work (*system.work.generate (key3.pub))
+					 .build ();
+		ASSERT_TRUE (nano::test::process (node_rep1, { send1, send2, send3, open1, open2, open3 }));
+		ASSERT_TRUE (nano::test::process (node_rep2, { send1, send2, send3, open1, open2, open3 }));
+	}
+
+	// Add rep keys to nodes
+	system.wallet (1)->insert_adhoc (key1.prv);
+	system.wallet (2)->insert_adhoc (key2.prv);
+	system.wallet (2)->insert_adhoc (key3.prv);
+
+	ASSERT_TIMELY_EQ (10s, node.online_reps.online (), weight_1 + weight_2 + weight_3);
+	ASSERT_ALWAYS_EQ (1s, node.online_reps.online (), weight_1 + weight_2 + weight_3);
+}
+
+// Online weight calculation should include local representative
+TEST (online_reps, observe_local)
+{
+	nano::test::system system;
+	auto & node = *system.add_node ();
+	ASSERT_EQ (0, node.online_reps.online ());
+	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
+	ASSERT_TIMELY_EQ (10s, node.online_reps.online (), nano::dev::constants.genesis_amount);
+	ASSERT_ALWAYS_EQ (1s, node.online_reps.online (), nano::dev::constants.genesis_amount);
+}
+
+// Online weight calculation should include slower but active representatives
+TEST (online_reps, observe_slow)
+{
+	nano::test::system system;
+	auto & node = *system.add_node ();
+	ASSERT_EQ (0, node.online_reps.online ());
+
+	// Enough to reach quorum by a single vote
+	auto const weight = nano::nano_ratio * 80000000;
+
+	nano::keypair key1, key2; // Fast and slow reps
+
+	// Distribute genesis voting weight
+	nano::block_builder builder;
+	auto send1 = builder.state ()
+				 .account (nano::dev::genesis_key.pub)
+				 .previous (nano::dev::genesis->hash ())
+				 .representative (nano::dev::genesis_key.pub)
+				 .balance (nano::dev::constants.genesis_amount - weight)
+				 .link (key1.pub)
+				 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
+				 .work (*system.work.generate (nano::dev::genesis->hash ()))
+				 .build ();
+	auto send2 = builder.state ()
+				 .account (nano::dev::genesis_key.pub)
+				 .previous (send1->hash ())
+				 .representative (nano::dev::genesis_key.pub)
+				 .balance (nano::dev::constants.genesis_amount - weight * 2)
+				 .link (key2.pub)
+				 .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
+				 .work (*system.work.generate (send1->hash ()))
+				 .build ();
+	auto open1 = builder.state ()
+				 .account (key1.pub)
+				 .previous (0)
+				 .representative (key1.pub)
+				 .balance (weight)
+				 .link (send1->hash ())
+				 .sign (key1.prv, key1.pub)
+				 .work (*system.work.generate (key1.pub))
+				 .build ();
+	auto open2 = builder.state ()
+				 .account (key2.pub)
+				 .previous (0)
+				 .representative (key2.pub)
+				 .balance (weight)
+				 .link (send2->hash ())
+				 .sign (key2.prv, key2.pub)
+				 .work (*system.work.generate (key2.pub))
+				 .build ();
+	ASSERT_TRUE (nano::test::process (node, { send1, send2, open1, open2 }));
+	nano::test::confirm (node, { send1, send2, open1, open2 });
+
+	ASSERT_EQ (node.active.size (), 0);
+
+	// Add a block that we can vote on
+	auto send_dummy = builder.state ()
+					  .account (nano::dev::genesis_key.pub)
+					  .previous (send2->hash ())
+					  .representative (nano::dev::genesis_key.pub)
+					  .balance (nano::dev::constants.genesis_amount - weight * 2 - nano::nano_ratio)
+					  .link (nano::keypair{}.pub)
+					  .sign (nano::dev::genesis_key.prv, nano::dev::genesis_key.pub)
+					  .work (*system.work.generate (send2->hash ()))
+					  .build ();
+	ASSERT_TRUE (nano::test::process (node, { send_dummy }));
+
+	// Wait for election for the block to be activated
+	std::shared_ptr<nano::election> election;
+	ASSERT_TIMELY (5s, election = node.active.election (send_dummy->qualified_root ()));
+	ASSERT_TRUE (election->contains (send_dummy->hash ()));
+
+	// Issue vote from a fast rep
+	auto vote_fast = nano::test::make_final_vote (key1, { send_dummy });
+	node.vote_processor.vote_blocking (vote_fast, nano::test::fake_channel (node));
+
+	ASSERT_TIMELY (5s, election->confirmed ());
+	ASSERT_TIMELY (5s, !node.active.active (send_dummy->qualified_root ())); // No longer present in AEC
+	ASSERT_TIMELY_EQ (5s, node.online_reps.online (), weight);
+
+	// Issue vote from a slow rep
+	auto vote_slow = nano::test::make_final_vote (key2, { send_dummy });
+	node.vote_processor.vote_blocking (vote_slow, nano::test::fake_channel (node));
+
+	// The slow rep weight should still be counted as online, even though it arrived slightly after the election already reached quorum
+	ASSERT_TIMELY_EQ (5s, node.online_reps.online (), weight * 2);
 }

nano/core_test/request_aggregator.cpp

@@ -28,7 +28,9 @@ TEST (request_aggregator, one)
 	nano::test::system system;
 	nano::node_config node_config = system.default_config ();
 	node_config.backlog_scan.enable = false;
-	auto & node (*system.add_node (node_config));
+	nano::node_flags node_flags;
+	node_flags.disable_rep_crawler = true;
+	auto & node (*system.add_node (node_config, node_flags));
 	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
 	nano::block_builder builder;
 	auto send1 = builder
@@ -77,7 +79,9 @@ TEST (request_aggregator, one_update)
 	nano::test::system system;
 	nano::node_config node_config = system.default_config ();
 	node_config.backlog_scan.enable = false;
-	auto & node (*system.add_node (node_config));
+	nano::node_flags node_flags;
+	node_flags.disable_rep_crawler = true;
+	auto & node (*system.add_node (node_config, node_flags));
 	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
 	nano::keypair key1;
 	auto send1 = nano::state_block_builder ()
@@ -143,7 +147,9 @@ TEST (request_aggregator, two)
 	nano::test::system system;
 	nano::node_config node_config = system.default_config ();
 	node_config.backlog_scan.enable = false;
-	auto & node (*system.add_node (node_config));
+	nano::node_flags node_flags;
+	node_flags.disable_rep_crawler = true;
+	auto & node (*system.add_node (node_config, node_flags));
 	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
 	nano::keypair key1;
 	nano::state_block_builder builder;
@@ -273,7 +279,9 @@ TEST (request_aggregator, split)
 	nano::test::system system;
 	nano::node_config node_config = system.default_config ();
 	node_config.backlog_scan.enable = false;
-	auto & node (*system.add_node (node_config));
+	nano::node_flags node_flags;
+	node_flags.disable_rep_crawler = true;
+	auto & node (*system.add_node (node_config, node_flags));
 	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
 	std::vector<std::pair<nano::block_hash, nano::root>> request;
 	std::vector<std::shared_ptr<nano::block>> blocks;
@@ -328,7 +336,9 @@ TEST (request_aggregator, channel_max_queue)
 	nano::node_config node_config = system.default_config ();
 	node_config.backlog_scan.enable = false;
 	node_config.request_aggregator.max_queue = 0;
-	auto & node (*system.add_node (node_config));
+	nano::node_flags node_flags;
+	node_flags.disable_rep_crawler = true;
+	auto & node (*system.add_node (node_config, node_flags));
 	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);
 	nano::block_builder builder;
 	auto send1 = builder
@@ -390,6 +400,7 @@ TEST (request_aggregator, cannot_vote)
 	nano::test::system system;
 	nano::node_flags flags;
 	flags.disable_request_loop = true;
+	flags.disable_rep_crawler = true;
 	auto & node (*system.add_node (flags));
 	nano::state_block_builder builder;
 	auto send1 = builder.make_block ()
@@ -548,7 +559,12 @@ TEST (request_aggregator, forked_open)
 TEST (request_aggregator, epoch_conflict)
 {
 	nano::test::system system;
-	auto & node = *system.add_node ();
+
+	nano::node_flags node_flags;
+	// Workaround for vote spacing dropping requests with the same root
+	// FIXME: Vote spacing should use full qualified root
+	node_flags.disable_rep_crawler = true;
+	auto & node = *system.add_node (node_flags);
 
 	// Voting needs a rep key set up on the node
 	system.wallet (0)->insert_adhoc (nano::dev::genesis_key.prv);

nano/core_test/vote_processor.cpp

@@ -114,7 +114,7 @@ TEST (vote_processor, weights)
 	auto & node (*system.nodes[0]);
 
 	// Create representatives of different weight levels
-	auto const stake = node.config.online_weight_minimum.number ();
+	auto const stake = nano::dev::genesis->balance ().number ();
 	auto const level0 = stake / 5000; // 0.02%
 	auto const level1 = stake / 500; // 0.2%
 	auto const level2 = stake / 50; // 2%

nano/lib/stats_enums.hpp

@@ -15,6 +15,7 @@ enum class type
 	test,
 	error,
 	message,
+	message_loopback,
 	block,
 	ledger,
 	ledger_notifications,
@@ -637,6 +638,7 @@ enum class detail
 	sample,
 	rep_new,
 	rep_update,
+	rep_trim,
 	update_online,
 
 	// error codes

nano/node/CMakeLists.txt

@@ -167,6 +167,8 @@ add_library(
   transport/fwd.hpp
   transport/inproc.hpp
   transport/inproc.cpp
+  transport/loopback.cpp
+  transport/loopback.hpp
   transport/message_deserializer.hpp
   transport/message_deserializer.cpp
   transport/tcp_channels.hpp

nano/node/active_elections.cpp

@@ -406,12 +406,14 @@ nano::election_insertion_result nano::active_elections::insert (std::shared_ptr<
 		if (!recently_confirmed.exists (root))
 		{
 			result.inserted = true;
-			auto observe_rep_cb = [&node = node] (auto const & rep_a) {
-				// TODO: Is this neccessary? Move this outside of the election class
-				// Representative is defined as online if replying to live votes or rep_crawler queries
+
+			// Passing this callback into the election is important
+			// We need to observe and update the online voting weight *before* election quorum is checked
+			auto observe_rep_callback = [&node = node] (auto const & rep_a) {
 				node.online_reps.observe (rep_a);
 			};
-			result.election = nano::make_shared<nano::election> (node, block_a, nullptr, observe_rep_cb, election_behavior_a);
+			result.election = nano::make_shared<nano::election> (node, block_a, nullptr, observe_rep_callback, election_behavior_a);
+
 			roots.get<tag_root> ().emplace (entry{ root, result.election, std::move (erased_callback_a) });
 			node.vote_router.connect (hash, result.election);