Remove representatives with closed channels (#2530)

nano/core_test/node.cpp

@@ -2269,6 +2269,27 @@ TEST (node, rep_remove)
 	ASSERT_EQ (node1->network.endpoint (), list[0]->get_endpoint ());
 }
 
+TEST (node, rep_connection_close)
+{
+	nano::system system (2);
+	auto & node1 (*system.nodes[0]);
+	auto & node2 (*system.nodes[1]);
+	// Add working representative (node 2)
+	system.wallet (1)->insert_adhoc (nano::test_genesis_key.prv);
+	system.deadline_set (10s);
+	while (node1.rep_crawler.representative_count () != 1)
+	{
+		ASSERT_NO_ERROR (system.poll ());
+	}
+	node2.stop ();
+	// Remove representative with closed channel
+	system.deadline_set (10s);
+	while (node1.rep_crawler.representative_count () != 0)
+	{
+		ASSERT_NO_ERROR (system.poll ());
+	}
+}
+
 // Test that nodes can disable representative voting
 TEST (node, no_voting)
 {

nano/node/node.cpp

@@ -164,6 +164,7 @@ startup_time (std::chrono::steady_clock::now ())
 			observers.wallet.notify (active);
 		};
 		network.channel_observer = [this](std::shared_ptr<nano::transport::channel> channel_a) {
+			assert (channel_a != nullptr);
 			observers.endpoint.notify (channel_a);
 		};
 		network.disconnect_observer = [this]() {

nano/node/repcrawler.cpp

@@ -44,6 +44,7 @@ void nano::rep_crawler::validate ()
 	{
 		auto & vote = i.second;
 		auto & channel = i.first;
+		assert (channel != nullptr);
 		nano::uint128_t rep_weight = node.ledger.weight (vote->account);
 		if (rep_weight > minimum)
 		{
@@ -165,6 +166,7 @@ void nano::rep_crawler::query (std::vector<std::shared_ptr<nano::transport::chan
 	}
 	for (auto i (channels_a.begin ()), n (channels_a.end ()); i != n; ++i)
 	{
+		assert (*i != nullptr);
 		on_rep_request (*i);
 		node.network.send_confirm_req (*i, block);
 	}
@@ -233,16 +235,18 @@ nano::uint128_t nano::rep_crawler::total_weight () const
 void nano::rep_crawler::on_rep_request (std::shared_ptr<nano::transport::channel> channel_a)
 {
 	nano::lock_guard<std::mutex> lock (probable_reps_mutex);
-
-	probably_rep_t::index<tag_channel_ref>::type & channel_ref_index = probable_reps.get<tag_channel_ref> ();
-
-	// Find and update the timestamp on all reps available on the endpoint (a single host may have multiple reps)
-	auto itr_pair = probable_reps.get<tag_channel_ref> ().equal_range (*channel_a);
-	for (; itr_pair.first != itr_pair.second; itr_pair.first++)
+	if (channel_a->get_tcp_endpoint ().address () != boost::asio::ip::address_v6::any ())
 	{
-		channel_ref_index.modify (itr_pair.first, [](nano::representative & value_a) {
-			value_a.last_request = std::chrono::steady_clock::now ();
-		});
+		probably_rep_t::index<tag_channel_ref>::type & channel_ref_index = probable_reps.get<tag_channel_ref> ();
+
+		// Find and update the timestamp on all reps available on the endpoint (a single host may have multiple reps)
+		auto itr_pair = channel_ref_index.equal_range (*channel_a);
+		for (; itr_pair.first != itr_pair.second; itr_pair.first++)
+		{
+			channel_ref_index.modify (itr_pair.first, [](nano::representative & value_a) {
+				value_a.last_request = std::chrono::steady_clock::now ();
+			});
+		}
 	}
 }
 
@@ -252,9 +256,19 @@ void nano::rep_crawler::cleanup_reps ()
 	{
 		// Check known rep channels
 		nano::lock_guard<std::mutex> lock (probable_reps_mutex);
-		for (auto const & rep : probable_reps.get<tag_last_request> ())
+		auto iterator (probable_reps.get<tag_last_request> ().begin ());
+		while (iterator != probable_reps.get<tag_last_request> ().end ())
 		{
-			channels.push_back (rep.channel);
+			if (iterator->channel->get_tcp_endpoint ().address () != boost::asio::ip::address_v6::any ())
+			{
+				channels.push_back (iterator->channel);
+				++iterator;
+			}
+			else
+			{
+				// Remove reps with closed channels
+				iterator = probable_reps.get<tag_last_request> ().erase (iterator);
+			}
 		}
 	}
 	// Remove reps with inactive channels

nano/node/repcrawler.hpp

@@ -31,6 +31,7 @@ public:
 	representative (nano::account account_a, nano::amount weight_a, std::shared_ptr<nano::transport::channel> channel_a) :
 	account (account_a), weight (weight_a), channel (channel_a)
 	{
+		assert (channel != nullptr);
 	}
 	std::reference_wrapper<nano::transport::channel const> channel_ref () const
 	{