Peer cache (#4574)

* Headers for forward declarations

* Conversions for endpoint_key

* Time helpers

* Store timestamps in peer store

* Introduce `peer_cache`

* Tests for peer cache

* Reach out to cached peers

* Test reconnecting to cached peers

* Fixes

* Rename to `peer_history`

nano/core_test/CMakeLists.txt

@@ -41,6 +41,7 @@ add_executable(
   processor_service.cpp
   rep_crawler.cpp
   receivable.cpp
+  peer_history.cpp
   peer_container.cpp
   rep_weight_store.cpp
   scheduler_buckets.cpp

nano/core_test/block_store.cpp

@@ -1157,28 +1157,31 @@ TEST (block_store, peers)
 		ASSERT_EQ (store->peer.count (transaction), 0);
 
 		// Add one
-		store->peer.put (transaction, endpoint);
+		store->peer.put (transaction, endpoint, 37);
 		ASSERT_TRUE (store->peer.exists (transaction, endpoint));
 	}
 
 	// Confirm that it can be found
 	{
 		auto transaction (store->tx_begin_read ());
 		ASSERT_EQ (store->peer.count (transaction), 1);
+		ASSERT_EQ (store->peer.get (transaction, endpoint), 37);
 	}
 
 	// Add another one and check that it (and the existing one) can be found
 	nano::endpoint_key endpoint1 (boost::asio::ip::address_v6::any ().to_bytes (), 101);
 	{
 		auto transaction (store->tx_begin_write ());
-		store->peer.put (transaction, endpoint1);
+		store->peer.put (transaction, endpoint1, 42);
 		ASSERT_TRUE (store->peer.exists (transaction, endpoint1)); // Check new peer is here
 		ASSERT_TRUE (store->peer.exists (transaction, endpoint)); // Check first peer is still here
 	}
 
 	{
 		auto transaction (store->tx_begin_read ());
 		ASSERT_EQ (store->peer.count (transaction), 2);
+		ASSERT_EQ (store->peer.get (transaction, endpoint), 37);
+		ASSERT_EQ (store->peer.get (transaction, endpoint1), 42);
 	}
 
 	// Delete the first one

nano/core_test/ledger.cpp

@@ -5439,7 +5439,7 @@ TEST (ledger, migrate_lmdb_to_rocksdb)
 		store.confirmation_height.put (transaction, nano::dev::genesis_key.pub, { 2, send->hash () });
 
 		store.online_weight.put (transaction, 100, nano::amount (2));
-		store.peer.put (transaction, endpoint_key);
+		store.peer.put (transaction, endpoint_key, 37);
 
 		store.pending.put (transaction, nano::pending_key (nano::dev::genesis_key.pub, send->hash ()), nano::pending_info (nano::dev::genesis_key.pub, 100, nano::epoch::epoch_0));
 		store.pruned.put (transaction, send->hash ());

nano/core_test/network.cpp

@@ -1044,6 +1044,57 @@ TEST (network, fill_keepalive_self)
 	ASSERT_EQ (target[2].port (), system.nodes[1]->network.port);
 }
 
+TEST (network, reconnect_cached)
+{
+	nano::test::system system;
+
+	nano::node_flags flags;
+	// Disable non realtime sockets
+	flags.disable_bootstrap_bulk_push_client = true;
+	flags.disable_bootstrap_bulk_pull_server = true;
+	flags.disable_bootstrap_listener = true;
+	flags.disable_lazy_bootstrap = true;
+	flags.disable_legacy_bootstrap = true;
+	flags.disable_wallet_bootstrap = true;
+
+	auto & node1 = *system.add_node (flags);
+	auto & node2 = *system.add_node (flags);
+
+	ASSERT_EQ (node1.network.size (), 1);
+	ASSERT_EQ (node2.network.size (), 1);
+
+	auto channels1 = node1.network.list ();
+	auto channels2 = node2.network.list ();
+	ASSERT_EQ (channels1.size (), 1);
+	ASSERT_EQ (channels2.size (), 1);
+	auto channel1 = channels1.front ();
+	auto channel2 = channels2.front ();
+
+	// Enusre current peers are cached
+	node1.peer_history.trigger ();
+	node2.peer_history.trigger ();
+	ASSERT_TIMELY_EQ (5s, node1.peer_history.size (), 1);
+	ASSERT_TIMELY_EQ (5s, node2.peer_history.size (), 1);
+
+	// Kill channels
+	channel1->close ();
+	channel2->close ();
+
+	auto channel_exists = [] (auto & node, auto & channel) {
+		auto channels = node.network.list ();
+		return std::find (channels.begin (), channels.end (), channel) != channels.end ();
+	};
+
+	ASSERT_TIMELY (5s, !channel_exists (node1, channel1));
+	ASSERT_TIMELY (5s, !channel_exists (node2, channel2));
+
+	// Peers should reconnect after a while
+	ASSERT_TIMELY_EQ (5s, node1.network.size (), 1);
+	ASSERT_TIMELY_EQ (5s, node2.network.size (), 1);
+	ASSERT_TRUE (node1.network.find_node_id (node2.node_id.pub));
+	ASSERT_TRUE (node2.network.find_node_id (node1.node_id.pub));
+}
+
 /*
  * Tests that channel and channel container removes channels with dead local sockets
  */

nano/core_test/node.cpp

@@ -2813,10 +2813,10 @@ TEST (node, peers)
 	{
 		// Add a peer to the database
 		auto transaction (store.tx_begin_write ());
-		store.peer.put (transaction, endpoint_key);
+		store.peer.put (transaction, endpoint_key, 37);
 
 		// Add a peer which is not contactable
-		store.peer.put (transaction, nano::endpoint_key{ boost::asio::ip::address_v6::any ().to_bytes (), 55555 });
+		store.peer.put (transaction, nano::endpoint_key{ boost::asio::ip::address_v6::any ().to_bytes (), 55555 }, 42);
 	}
 
 	node2->start ();
@@ -2845,7 +2845,7 @@ TEST (node, peers)
 	ASSERT_TIMELY (10s, node2->network.empty ());
 }
 
-TEST (node, peer_cache_restart)
+TEST (node, peer_history_restart)
 {
 	nano::test::system system (1);
 	auto node1 (system.nodes[0]);
@@ -2860,7 +2860,7 @@ TEST (node, peer_cache_restart)
 		{
 			// Add a peer to the database
 			auto transaction (store.tx_begin_write ());
-			store.peer.put (transaction, endpoint_key);
+			store.peer.put (transaction, endpoint_key, 37);
 		}
 		node2->start ();
 		ASSERT_TIMELY (10s, !node2->network.empty ());

nano/core_test/peer_history.cpp

@@ -0,0 +1,48 @@
+#include <nano/node/peer_history.hpp>
+#include <nano/test_common/system.hpp>
+#include <nano/test_common/testutil.hpp>
+
+#include <gtest/gtest.h>
+
+TEST (peer_history, store_live)
+{
+	nano::test::system system;
+
+	auto & node1 = *system.add_node ();
+	auto & node2 = *system.add_node ();
+	auto & node3 = *system.add_node ();
+
+	ASSERT_TIMELY (5s, node1.peer_history.exists (node2.network.endpoint ()));
+	ASSERT_TIMELY (5s, node1.peer_history.exists (node3.network.endpoint ()));
+
+	ASSERT_TIMELY (5s, node2.peer_history.exists (node1.network.endpoint ()));
+	ASSERT_TIMELY (5s, node2.peer_history.exists (node3.network.endpoint ()));
+
+	ASSERT_TIMELY (5s, node3.peer_history.exists (node1.network.endpoint ()));
+	ASSERT_TIMELY (5s, node3.peer_history.exists (node2.network.endpoint ()));
+}
+
+TEST (peer_history, erase_old)
+{
+	nano::test::system system;
+
+	auto & node1 = *system.add_node ();
+	auto & node2 = *system.add_node ();
+
+	ASSERT_TIMELY (5s, node1.peer_history.exists (node2.network.endpoint ()));
+	ASSERT_TIMELY (5s, node2.peer_history.exists (node1.network.endpoint ()));
+
+	// Endpoint won't be available after node is stopped
+	auto node2_endpoint = node2.network.endpoint ();
+
+	system.stop_node (node2);
+
+	auto cached1 = node1.peer_history.peers ();
+	ASSERT_EQ (cached1.size (), 1);
+	ASSERT_EQ (cached1[0], node2_endpoint);
+
+	ASSERT_TIMELY (5s, !node1.peer_history.exists (node2_endpoint));
+
+	auto cached2 = node1.peer_history.peers ();
+	ASSERT_EQ (cached2.size (), 0);
+}

nano/lib/common.hpp

@@ -0,0 +1,10 @@
+#pragma once
+
+#include <nano/boost/asio/ip/tcp.hpp>
+#include <nano/boost/asio/ip/udp.hpp>
+
+namespace nano
+{
+using endpoint = boost::asio::ip::udp::endpoint;
+using tcp_endpoint = boost::asio::ip::tcp::endpoint;
+}

nano/lib/logging_enums.hpp

@@ -76,6 +76,7 @@ enum class type
 	syn_cookies,
 	thread_runner,
 	signal_manager,
+	peer_history,
 
 	// bootstrap
 	bulk_pull_client,

nano/lib/stats_enums.hpp

@@ -57,6 +57,7 @@ enum class type : uint8_t
 	local_block_broadcaster,
 	rep_tiers,
 	syn_cookies,
+	peer_history,
 
 	bootstrap_ascending,
 	bootstrap_ascending_accounts,
@@ -79,6 +80,8 @@ enum class detail : uint8_t
 	ignored,
 	update,
 	updated,
+	inserted,
+	erased,
 	request,
 	broadcast,
 	cleanup,
@@ -226,7 +229,10 @@ enum class detail : uint8_t
 	// network
 	loop_keepalive,
 	loop_reachout,
+	loop_reachout_cached,
 	merge_peer,
+	reachout_live,
+	reachout_cached,
 
 	// tcp
 	tcp_write_drop,

nano/lib/thread_roles.cpp

@@ -131,6 +131,9 @@ std::string nano::thread_role::get_string (nano::thread_role::name role)
 		case nano::thread_role::name::tcp_listener:
 			thread_role_name_string = "TCP listener";
 			break;
+		case nano::thread_role::name::peer_history:
+			thread_role_name_string = "Peer history";
+			break;
 		default:
 			debug_assert (false && "nano::thread_role::get_string unhandled thread role");
 	}

nano/lib/thread_roles.hpp

@@ -50,6 +50,7 @@ enum class name
 	network_reachout,
 	signal_manager,
 	tcp_listener,
+	peer_history,
 };
 
 std::string_view to_string (name);

nano/lib/timer.hpp

@@ -105,13 +105,23 @@ inline millis_t milliseconds_since_epoch ()
 	return std::chrono::duration_cast<std::chrono::milliseconds> (std::chrono::system_clock::now ().time_since_epoch ()).count ();
 }
 
+inline std::chrono::time_point<std::chrono::system_clock> from_milliseconds_since_epoch (nano::millis_t millis)
+{
+	return std::chrono::time_point<std::chrono::system_clock> (std::chrono::milliseconds{ millis });
+}
+
 using seconds_t = uint64_t;
 
 inline seconds_t seconds_since_epoch ()
 {
 	return std::chrono::duration_cast<std::chrono::seconds> (std::chrono::system_clock::now ().time_since_epoch ()).count ();
 }
 
+inline std::chrono::time_point<std::chrono::system_clock> from_seconds_since_epoch (nano::seconds_t seconds)
+{
+	return std::chrono::time_point<std::chrono::system_clock> (std::chrono::seconds{ seconds });
+}
+
 inline nano::millis_t time_difference (nano::millis_t start, nano::millis_t end)
 {
 	return end > start ? (end - start) : 0;

nano/nano_node/entry.cpp

@@ -1856,11 +1856,10 @@ int main (int argc, char * const * argv)
 		{
 			auto inactive_node = nano::default_inactive_node (data_path, vm);
 			auto node = inactive_node->node;
-			auto transaction (node->store.tx_begin_read ());
-
-			for (auto i (node->store.peer.begin (transaction)), n (node->store.peer.end ()); i != n; ++i)
+			auto peers = node->peer_history.peers ();
+			for (auto const & peer : peers)
 			{
-				std::cout << boost::str (boost::format ("%1%\n") % nano::endpoint (boost::asio::ip::address_v6 (i->first.address_bytes ()), i->first.port ()));
+				std::cout << peer << std::endl;
 			}
 		}
 		else if (vm.count ("debug_cemented_block_count"))

nano/node/CMakeLists.txt

@@ -119,6 +119,8 @@ add_library(
   openclconfig.cpp
   openclwork.hpp
   openclwork.cpp
+  peer_history.hpp
+  peer_history.cpp
   peer_exclusion.hpp
   peer_exclusion.cpp
   portmapping.hpp

nano/node/common.hpp

@@ -12,9 +12,6 @@
 
 namespace nano
 {
-using endpoint = boost::asio::ip::udp::endpoint;
-using tcp_endpoint = boost::asio::ip::tcp::endpoint;
-
 bool parse_port (std::string const &, uint16_t &);
 bool parse_address (std::string const &, boost::asio::ip::address &);
 bool parse_address_port (std::string const &, boost::asio::ip::address &, uint16_t &);

nano/node/fwd.hpp

@@ -0,0 +1,11 @@
+#pragma once
+
+#include <nano/store/fwd.hpp>
+
+namespace nano
+{
+class logger;
+class node;
+class network;
+class stats;
+}