Update confirmation height in another thread

nano/core_test/network.cpp

@@ -6,6 +6,11 @@
 
 using namespace std::chrono_literals;
 
+namespace
+{
+void prevent_frontier_confirmation_height_checking (std::vector<std::shared_ptr<nano::node>> & nodes);
+}
+
 TEST (network, tcp_connection)
 {
 	boost::asio::io_context io_ctx;
@@ -1455,8 +1460,8 @@ TEST (confirmation_height, single)
 
 TEST (confirmation_height, multiple)
 {
-	auto amount (std::numeric_limits<nano::uint128_t>::max ());
 	nano::system system (24000, 2);
+	prevent_frontier_confirmation_height_checking (system.nodes);
 	nano::keypair key1;
 	nano::keypair key2;
 	nano::keypair key3;
@@ -1638,9 +1643,9 @@ TEST (confirmation_height, gap_bootstrap)
 TEST (confirmation_height, gap_live)
 {
 	nano::system system (24000, 2);
+	prevent_frontier_confirmation_height_checking (system.nodes);
 	nano::keypair destination;
 	system.wallet (0)->insert_adhoc (nano::test_genesis_key.prv);
-	nano::block_hash latest1 (system.nodes[0]->latest (nano::test_genesis_key.pub));
 	system.wallet (1)->insert_adhoc (destination.prv);
 
 	nano::genesis genesis;
@@ -1687,7 +1692,7 @@ TEST (confirmation_height, gap_live)
 		while (true)
 		{
 			auto transaction = node->store.tx_begin_read ();
-			if (node->ledger.block_confirmed (transaction, open1->hash ()))
+			if (node->ledger.block_confirmed (transaction, receive2->hash ()))
 			{
 				break;
 			}
@@ -1696,19 +1701,15 @@ TEST (confirmation_height, gap_live)
 		}
 
 		// This should confirm the open block and the source of the receive blocks
-		{
-			auto transaction (node->store.tx_begin ());
-			auto unchecked_count (node->store.unchecked_count (transaction));
-			ASSERT_EQ (unchecked_count, 0);
+		auto transaction (node->store.tx_begin ());
+		auto unchecked_count (node->store.unchecked_count (transaction));
+		ASSERT_EQ (unchecked_count, 0);
 
-			nano::account_info account_info;
-			ASSERT_FALSE (node->store.account_get (transaction, nano::test_genesis_key.pub, account_info));
-			ASSERT_EQ (4, account_info.block_count);
-			ASSERT_EQ (2, account_info.confirmation_height);
-			ASSERT_FALSE (node->store.account_get (transaction, destination.pub, account_info));
-			ASSERT_EQ (1, account_info.confirmation_height);
-			ASSERT_EQ (3, account_info.block_count);
-		}
+		nano::account_info account_info;
+		ASSERT_FALSE (node->store.account_get (transaction, nano::test_genesis_key.pub, account_info));
+		ASSERT_EQ (4, account_info.confirmation_height);
+		ASSERT_FALSE (node->store.account_get (transaction, destination.pub, account_info));
+		ASSERT_EQ (3, account_info.confirmation_height);
 	}
 }
 
@@ -1775,3 +1776,13 @@ TEST (bootstrap, tcp_listener_timeout_keepalive)
 		ASSERT_NO_ERROR (system.poll ());
 	}
 }
+
+namespace
+{
+void prevent_frontier_confirmation_height_checking (std::vector<std::shared_ptr<nano::node>> & nodes)
+{
+	auto unreachable_time = std::chrono::steady_clock::now () + std::chrono::seconds (10000);
+	nodes.front ()->active.next_frontier_check = unreachable_time;
+	nodes.back ()->active.next_frontier_check = unreachable_time;
+}
+}

nano/lib/utility.cpp

@@ -96,6 +96,8 @@ namespace thread_role
 			case nano::thread_role::name::signature_checking:
 				thread_role_name_string = "Signature check";
 				break;
+			case nano::thread_role::name::confirmation_height_processing:
+				thread_role_name_string = "Conf height";
 		}
 
 		/*

nano/lib/utility.hpp

@@ -85,7 +85,8 @@ namespace thread_role
 		wallet_actions,
 		bootstrap_initiator,
 		voting,
-		signature_checking
+		signature_checking,
+		confirmation_height_processing
 	};
 	/*
 	 * Get/Set the identifier for the current thread

nano/node/CMakeLists.txt

@@ -21,6 +21,8 @@ add_library (node
 	cli.cpp
 	common.cpp
 	common.hpp
+	confirmationheightprocessor.hpp
+	confirmationheightprocessor.cpp
 	daemonconfig.hpp
 	daemonconfig.cpp
 	ipc.hpp

nano/node/confirmationheightprocessor.cpp

@@ -0,0 +1,258 @@
+#include <boost/optional.hpp>
+#include <boost/polymorphic_pointer_cast.hpp>
+#include <nano/lib/numbers.hpp>
+#include <nano/lib/timer.hpp>
+#include <nano/lib/utility.hpp>
+#include <nano/node/confirmationheightprocessor.hpp>
+#include <nano/node/node.hpp> // For active_transactions
+#include <nano/secure/blockstore.hpp>
+#include <nano/secure/common.hpp>
+#include <nano/secure/ledger.hpp>
+
+constexpr std::chrono::milliseconds nano::confirmation_height_processor::batch_write_delta;
+
+nano::confirmation_height_processor::confirmation_height_processor (nano::block_store & store, nano::ledger & ledger, nano::active_transactions & active, nano::logger_mt & logger) :
+store (store),
+ledger (ledger),
+active (active),
+logger (logger),
+thread ([this]() {
+	nano::thread_role::set (nano::thread_role::name::confirmation_height_processing);
+	this->run ();
+})
+{
+}
+
+nano::confirmation_height_processor::~confirmation_height_processor ()
+{
+	stop ();
+}
+
+void nano::confirmation_height_processor::stop ()
+{
+	{
+		std::lock_guard<std::mutex> lock (mutex);
+		stopped = true;
+	}
+	condition.notify_one ();
+	if (thread.joinable ())
+	{
+		thread.join ();
+	}
+}
+
+void nano::confirmation_height_processor::run ()
+{
+	std::unique_lock<std::mutex> lk (mutex);
+	while (!stopped)
+	{
+		if (!pending_confirmations.empty ())
+		{
+			auto pending_confirmation = pending_confirmations.front ();
+			pending_confirmations.pop ();
+			lk.unlock ();
+			add_confirmation_height (pending_confirmation);
+			lk.lock ();
+		}
+		else
+		{
+			condition.wait (lk);
+		}
+	}
+}
+
+void nano::confirmation_height_processor::add (nano::block_hash const & hash_a)
+{
+	{
+		std::lock_guard<std::mutex> lk (mutex);
+		pending_confirmations.push (hash_a);
+	}
+	condition.notify_one ();
+}
+
+/**
+ * For all the blocks below this height which have been implicitly confirmed check if they
+ * are open/receive blocks, and if so follow the source blocks and iteratively repeat to genesis.
+ * To limit write locking and to keep the confirmation height ledger correctly synced, confirmations are
+ * written from the ground upwards in batches.
+ */
+void nano::confirmation_height_processor::add_confirmation_height (nano::block_hash const & hash_a)
+{
+	std::stack<open_receive_source_pair> open_receive_source_pairs;
+	boost::optional<conf_height_details> open_receive_details;
+	auto current = hash_a;
+	nano::account_info account_info;
+	nano::genesis genesis;
+	auto genesis_hash = genesis.hash ();
+	std::queue<conf_height_details> pending;
+
+	nano::timer<std::chrono::milliseconds> timer;
+	timer.start ();
+
+	do
+	{
+		if (!open_receive_source_pairs.empty ())
+		{
+			open_receive_details = open_receive_source_pairs.top ().open_receive_details;
+			current = open_receive_source_pairs.top ().source_hash;
+		}
+
+		auto transaction (store.tx_begin_read ());
+		auto block_height = (store.block_account_height (transaction, current));
+		nano::account account (ledger.account (transaction, current));
+		release_assert (!store.account_get (transaction, account, account_info));
+		auto confirmation_height = account_info.confirmation_height;
+		auto count_before_open_receive = open_receive_source_pairs.size ();
+
+		auto hash (current);
+		if (block_height > confirmation_height)
+		{
+			collect_unconfirmed_receive_and_sources_for_account (block_height, confirmation_height, current, genesis_hash, open_receive_source_pairs, account, transaction);
+		}
+
+		// If this adds no more open_receive blocks, then we can now confirm this account as well as the linked open/receive block
+		// Collect as pending any writes to the database and do them in bulk after a certain time.
+		auto confirmed_receives_pending = (count_before_open_receive != open_receive_source_pairs.size ());
+		if (!confirmed_receives_pending)
+		{
+			if (block_height > confirmation_height)
+			{
+				pending.emplace (account, hash, block_height);
+			}
+
+			if (open_receive_details)
+			{
+				pending.push (*open_receive_details);
+			}
+
+			if (!open_receive_source_pairs.empty ())
+			{
+				open_receive_source_pairs.pop ();
+			}
+		}
+
+		// Check whether writing to the database should be done now
+		if ((timer.after_deadline (batch_write_delta) || open_receive_source_pairs.empty ()) && !pending.empty ())
+		{
+			auto error = write_pending (pending);
+
+			// Don't set any more blocks as confirmed from the originally hash if an inconsistency is found
+			if (error)
+			{
+				break;
+			}
+			assert (pending.empty ());
+			timer.restart ();
+		}
+	} while (!open_receive_source_pairs.empty ());
+}
+
+// Returns true if there was an error in finding one of the blocks to write the confirmation height for.
+bool nano::confirmation_height_processor::write_pending (std::queue<conf_height_details> & all_pending)
+{
+	nano::account_info account_info;
+	auto transaction (store.tx_begin_write ());
+	while (!all_pending.empty ())
+	{
+		const auto & pending = all_pending.front ();
+		auto error = store.account_get (transaction, pending.account, account_info);
+		release_assert (!error);
+		if (pending.height > account_info.confirmation_height)
+		{
+#ifdef NDEBUG
+			auto block = store.block_get (transaction, pending.hash);
+#else
+			// Do more thorough checking in Debug mode
+			nano::block_sideband sideband;
+			auto block = store.block_get (transaction, pending.hash, &sideband);
+			assert (block != nullptr);
+			assert (sideband.height == pending.height);
+#endif
+			// Check that the block still exists as there may have been changes outside this processor.
+			if (!block)
+			{
+				logger.always_log ("Failed to write confirmation height for: ", pending.hash.to_string ());
+				ledger.stats.inc (nano::stat::type::confirmation_height, nano::stat::detail::invalid_block);
+				return true;
+			}
+
+			ledger.stats.add (nano::stat::type::confirmation_height, nano::stat::detail::blocks_confirmed, nano::stat::dir::in, pending.height - account_info.confirmation_height);
+			account_info.confirmation_height = pending.height;
+			store.account_put (transaction, pending.account, account_info);
+		}
+		all_pending.pop ();
+	}
+	return false;
+}
+
+void nano::confirmation_height_processor::collect_unconfirmed_receive_and_sources_for_account (uint64_t block_height, uint64_t confirmation_height, nano::block_hash & current, const nano::block_hash & genesis_hash, std::stack<open_receive_source_pair> & open_receive_source_pairs, nano::account const & account, nano::transaction & transaction)
+{
+	// Get the last confirmed block in this account chain
+	auto num_to_confirm = block_height - confirmation_height;
+	while (num_to_confirm > 0 && !current.is_zero ())
+	{
+		active.confirm_block (current);
+		nano::block_sideband sideband;
+		auto block (store.block_get (transaction, current, &sideband));
+		if (block)
+		{
+			if (block->type () == nano::block_type::receive || (block->type () == nano::block_type::open && block->hash () != genesis_hash))
+			{
+				open_receive_source_pairs.emplace (conf_height_details{ account, current, sideband.height }, block->source ());
+			}
+			else
+			{
+				// Then check state blocks
+				auto state = boost::dynamic_pointer_cast<nano::state_block> (block);
+				if (state != nullptr)
+				{
+					nano::block_hash previous (state->hashables.previous);
+					if (!previous.is_zero ())
+					{
+						if (state->hashables.balance.number () >= ledger.balance (transaction, previous) && !state->hashables.link.is_zero () && !ledger.is_epoch_link (state->hashables.link))
+						{
+							open_receive_source_pairs.emplace (conf_height_details{ account, current, sideband.height }, state->hashables.link);
+						}
+					}
+					// State open blocks are always receive or epoch
+					else if (!ledger.is_epoch_link (state->hashables.link))
+					{
+						open_receive_source_pairs.emplace (conf_height_details{ account, current, sideband.height }, state->hashables.link);
+					}
+				}
+			}
+			current = block->previous ();
+		}
+		--num_to_confirm;
+	}
+}
+
+size_t nano::confirmation_height_processor::pending_confirmations_size ()
+{
+	std::lock_guard<std::mutex> lk (mutex);
+	return pending_confirmations.size ();
+}
+
+namespace nano
+{
+confirmation_height_processor::conf_height_details::conf_height_details (nano::account const & account_a, nano::block_hash const & hash_a, uint64_t height_a) :
+account (account_a),
+hash (hash_a),
+height (height_a)
+{
+}
+
+confirmation_height_processor::open_receive_source_pair::open_receive_source_pair (confirmation_height_processor::conf_height_details const & open_receive_details_a, const block_hash & source_a) :
+open_receive_details (open_receive_details_a),
+source_hash (source_a)
+{
+}
+
+std::unique_ptr<seq_con_info_component> collect_seq_con_info (confirmation_height_processor & confirmation_height_processor, const std::string & name)
+{
+	size_t pending_confirmation_count = confirmation_height_processor.pending_confirmations_size ();
+	auto composite = std::make_unique<seq_con_info_composite> (name);
+	composite->add_component (std::make_unique<seq_con_info_leaf> (seq_con_info{ "pending_confirmations", pending_confirmation_count, sizeof (nano::block_hash) }));
+	return composite;
+}
+}