Some confirmed block observer callbacks being missed

nano/core_test/confirmation_height.cpp

@@ -842,50 +842,6 @@ TEST (confirmation_height, modified_chain)
 
 namespace nano
 {
-TEST (confirmation_height, pending_observer_callbacks)
-{
-	auto test_mode = [](nano::confirmation_height_mode mode_a) {
-		nano::system system;
-		nano::node_flags node_flags;
-		node_flags.confirmation_height_processor_mode = mode_a;
-		nano::node_config node_config (nano::get_available_port (), system.logging);
-		node_config.frontiers_confirmation = nano::frontiers_confirmation_mode::disabled;
-		auto node = system.add_node (node_config, node_flags);
-
-		system.wallet (0)->insert_adhoc (nano::test_genesis_key.prv);
-		nano::block_hash latest (node->latest (nano::test_genesis_key.pub));
-
-		nano::keypair key1;
-		nano::send_block send (latest, key1.pub, nano::genesis_amount - nano::Gxrb_ratio, nano::test_genesis_key.prv, nano::test_genesis_key.pub, *system.work.generate (latest));
-		auto send1 = std::make_shared<nano::send_block> (send.hash (), key1.pub, nano::genesis_amount - nano::Gxrb_ratio * 2, nano::test_genesis_key.prv, nano::test_genesis_key.pub, *system.work.generate (send.hash ()));
-
-		{
-			auto transaction = node->store.tx_begin_write ();
-			ASSERT_EQ (nano::process_result::progress, node->ledger.process (transaction, send).code);
-			ASSERT_EQ (nano::process_result::progress, node->ledger.process (transaction, *send1).code);
-		}
-
-		add_callback_stats (*node);
-
-		node->confirmation_height_processor.add (send1->hash ());
-
-		system.deadline_set (10s);
-		// Confirm the callback is not called under this circumstance because there is no election information
-		while (node->stats.count (nano::stat::type::http_callback, nano::stat::detail::http_callback, nano::stat::dir::out) != 1 || node->ledger.stats.count (nano::stat::type::observer, nano::stat::detail::all, nano::stat::dir::out) != 1)
-		{
-			ASSERT_NO_ERROR (system.poll ());
-		}
-
-		ASSERT_EQ (2, node->stats.count (nano::stat::type::confirmation_height, nano::stat::detail::blocks_confirmed, nano::stat::dir::in));
-		ASSERT_EQ (2, node->stats.count (nano::stat::type::confirmation_height, get_stats_detail (mode_a), nano::stat::dir::in));
-		ASSERT_EQ (3, node->ledger.cache.cemented_count);
-		ASSERT_EQ (0, node->active.election_winner_details_size ());
-	};
-
-	test_mode (nano::confirmation_height_mode::bounded);
-	test_mode (nano::confirmation_height_mode::unbounded);
-}
-
 TEST (confirmation_height, prioritize_frontiers)
 {
 	auto test_mode = [](nano::confirmation_height_mode mode_a) {

nano/node/active_transactions.cpp

@@ -950,22 +950,36 @@ boost::optional<nano::election_status_type> nano::active_transactions::confirm_b
 	auto hash (block_a->hash ());
 	nano::unique_lock<std::mutex> lock (mutex);
 	auto existing (blocks.find (hash));
+	boost::optional<nano::election_status_type> status_type;
 	if (existing != blocks.end ())
 	{
-		if (!existing->second->confirmed () && existing->second->status.winner->hash () == hash)
+		if (existing->second->status.winner && existing->second->status.winner->hash () == hash)
 		{
-			existing->second->confirm_once (nano::election_status_type::active_confirmation_height);
-			return nano::election_status_type::active_confirmation_height;
+			if (!existing->second->confirmed ())
+			{
+				existing->second->confirm_once (nano::election_status_type::active_confirmation_height);
+				status_type = nano::election_status_type::active_confirmation_height;
+			}
+			else
+			{
+#ifndef NDEBUG
+				nano::unique_lock<std::mutex> election_winners_lk (election_winner_details_mutex);
+				debug_assert (election_winner_details.find (hash) != election_winner_details.cend ());
+#endif
+				status_type = nano::election_status_type::active_confirmed_quorum;
+			}
 		}
 		else
 		{
-			return boost::optional<nano::election_status_type>{};
+			status_type = boost::optional<nano::election_status_type>{};
 		}
 	}
 	else
 	{
-		return nano::election_status_type::inactive_confirmation_height;
+		status_type = nano::election_status_type::inactive_confirmation_height;
 	}
+
+	return status_type;
 }
 
 size_t nano::active_transactions::priority_cementable_frontiers_size ()

nano/node/confirmation_height_bounded.cpp

@@ -491,6 +491,11 @@ bool nano::confirmation_height_bounded::cement_blocks (nano::write_guard & scope
 		scoped_write_guard_a.release ();
 		notify_observers_callback (cemented_blocks);
 	}
+	// Tests should check this already at the end, but not all blocks may have elections (e.g from manual calls to confirmation_height_processor::add), this should catch any inconsistencies on live/beta though
+	if (!network_params.network.is_test_network ())
+	{
+		debug_assert (ledger.stats.count (nano::stat::type::confirmation_height, nano::stat::detail::blocks_confirmed) == ledger.stats.count (nano::stat::type::observer, nano::stat::detail::all, nano::stat::dir::out));
+	}
 
 	debug_assert (pending_writes.empty ());
 	debug_assert (pending_writes_size == 0);
@@ -502,7 +507,7 @@ bool nano::confirmation_height_bounded::pending_empty () const
 	return pending_writes.empty ();
 }
 
-void nano::confirmation_height_bounded::prepare_new ()
+void nano::confirmation_height_bounded::reset ()
 {
 	accounts_confirmed_info.clear ();
 	accounts_confirmed_info_size = 0;

nano/node/confirmation_height_bounded.hpp

@@ -19,7 +19,7 @@ class confirmation_height_bounded final
 public:
 	confirmation_height_bounded (nano::ledger &, nano::write_database_queue &, std::chrono::milliseconds, nano::logger_mt &, std::atomic<bool> &, nano::block_hash const &, uint64_t &, std::function<void(std::vector<std::shared_ptr<nano::block>> const &)> const &, std::function<void(nano::block_hash const &)> const &, std::function<uint64_t ()> const &);
 	bool pending_empty () const;
-	void prepare_new ();
+	void reset ();
 	void process ();
 	bool cement_blocks (nano::write_guard & scoped_write_guard_a);
 

nano/node/confirmation_height_processor.cpp

@@ -73,7 +73,7 @@ void nano::confirmation_height_processor::run (confirmation_height_mode mode_a)
 				debug_assert (confirmation_height_bounded_processor.pending_empty ());
 				if (confirmation_height_unbounded_processor.pending_empty ())
 				{
-					confirmation_height_unbounded_processor.prepare_new ();
+					confirmation_height_unbounded_processor.reset ();
 				}
 				confirmation_height_unbounded_processor.process ();
 			}
@@ -83,7 +83,7 @@ void nano::confirmation_height_processor::run (confirmation_height_mode mode_a)
 				debug_assert (confirmation_height_unbounded_processor.pending_empty ());
 				if (confirmation_height_bounded_processor.pending_empty ())
 				{
-					confirmation_height_bounded_processor.prepare_new ();
+					confirmation_height_bounded_processor.reset ();
 				}
 				confirmation_height_bounded_processor.process ();
 			}
@@ -111,6 +111,7 @@ void nano::confirmation_height_processor::run (confirmation_height_mode mode_a)
 						confirmation_height_bounded_processor.cement_blocks (scoped_write_guard);
 					}
 					lock_and_cleanup ();
+					confirmation_height_bounded_processor.reset ();
 				}
 				else if (!confirmation_height_unbounded_processor.pending_empty ())
 				{
@@ -120,6 +121,7 @@ void nano::confirmation_height_processor::run (confirmation_height_mode mode_a)
 						confirmation_height_unbounded_processor.cement_blocks (scoped_write_guard);
 					}
 					lock_and_cleanup ();
+					confirmation_height_unbounded_processor.reset ();
 				}
 				else
 				{

nano/node/confirmation_height_unbounded.cpp

@@ -381,6 +381,11 @@ bool nano::confirmation_height_unbounded::cement_blocks (nano::write_guard & sco
 	scoped_write_guard_a.release ();
 	notify_observers_callback (cemented_blocks);
 
+	// Tests should check this already at the end, but not all blocks may have elections (e.g from manual calls to confirmation_height_processor::add), this should catch any inconsistencies on live/beta though
+	if (!network_params.network.is_test_network ())
+	{
+		debug_assert (ledger.stats.count (nano::stat::type::confirmation_height, nano::stat::detail::blocks_confirmed) == ledger.stats.count (nano::stat::type::observer, nano::stat::detail::all, nano::stat::dir::out));
+	}
 	debug_assert (pending_writes.empty ());
 	return false;
 }
@@ -406,7 +411,7 @@ bool nano::confirmation_height_unbounded::pending_empty () const
 	return pending_writes.empty ();
 }
 
-void nano::confirmation_height_unbounded::prepare_new ()
+void nano::confirmation_height_unbounded::reset ()
 {
 	// Separate blocks which are pending confirmation height can be batched by a minimum processing time (to improve lmdb disk write performance),
 	// so make sure the slate is clean when a new batch is starting.

nano/node/confirmation_height_unbounded.hpp

@@ -20,7 +20,7 @@ class confirmation_height_unbounded final
 public:
 	confirmation_height_unbounded (nano::ledger &, nano::write_database_queue &, std::chrono::milliseconds, nano::logger_mt &, std::atomic<bool> &, nano::block_hash const &, uint64_t &, std::function<void(std::vector<std::shared_ptr<nano::block>> const &)> const &, std::function<void(nano::block_hash const &)> const &, std::function<uint64_t ()> const &);
 	bool pending_empty () const;
-	void prepare_new ();
+	void reset ();
 	void process ();
 	bool cement_blocks (nano::write_guard &);
 
@@ -91,6 +91,7 @@ class confirmation_height_unbounded final
 	void collect_unconfirmed_receive_and_sources_for_account (uint64_t, uint64_t, nano::block_hash const &, nano::account const &, nano::read_transaction const &, std::vector<receive_source_pair> &, std::vector<nano::block_hash> &);
 	void prepare_iterated_blocks_for_cementing (preparation_data &);
 
+	nano::network_params network_params;
 	nano::ledger & ledger;
 	nano::write_database_queue & write_database_queue;
 	std::chrono::milliseconds batch_separate_pending_min_time;

nano/node/election.cpp

@@ -42,7 +42,9 @@ prioritized_m (prioritized_a)
 void nano::election::confirm_once (nano::election_status_type type_a)
 {
 	debug_assert (!node.active.mutex.try_lock ());
-	if (state_m.exchange (nano::election::state_t::confirmed) != nano::election::state_t::confirmed)
+	// This must be kept above the setting of election state, as dependent confirmed elections require up to date changes to election_winner_details
+	nano::unique_lock<std::mutex> election_winners_lk (node.active.election_winner_details_mutex);
+	if (state_m.exchange (nano::election::state_t::confirmed) != nano::election::state_t::confirmed && node.active.election_winner_details.find (status.winner->hash ()) == node.active.election_winner_details.cend ())
 	{
 		status.election_end = std::chrono::duration_cast<std::chrono::milliseconds> (std::chrono::system_clock::now ().time_since_epoch ());
 		status.election_duration = std::chrono::duration_cast<std::chrono::milliseconds> (std::chrono::steady_clock::now () - election_start);
@@ -54,15 +56,10 @@ void nano::election::confirm_once (nano::election_status_type type_a)
 		auto node_l (node.shared ());
 		auto confirmation_action_l (confirmation_action);
 		auto this_l = shared_from_this ();
-		if (status_l.type == nano::election_status_type::active_confirmation_height)
-		{
-			// Need to add dependent election results here (and not in process_confirmed which is called asynchronously) so that
-			// election winner details can be cleared consistently sucessfully in active_transactions::block_cemented_callback
-			node.active.add_election_winner_details (status_l.winner->hash (), this_l);
-		}
+		node.active.election_winner_details.emplace (status.winner->hash (), this_l);
 		node.active.add_recently_confirmed (status_l.winner->qualified_root (), status_l.winner->hash ());
+		node_l->process_confirmed (status_l, this_l);
 		node.background ([node_l, status_l, confirmation_action_l, this_l]() {
-			node_l->process_confirmed (status_l, this_l);
 			confirmation_action_l (status_l.winner);
 		});
 		adjust_dependent_difficulty ();

nano/node/node.cpp

@@ -1127,7 +1127,7 @@ void nano::node::block_confirm (std::shared_ptr<nano::block> block_a)
 
 bool nano::node::block_confirmed_or_being_confirmed (nano::transaction const & transaction_a, nano::block_hash const & hash_a)
 {
-	return ledger.block_confirmed (transaction_a, hash_a) || confirmation_height_processor.is_processing_block (hash_a);
+	return confirmation_height_processor.is_processing_block (hash_a) || ledger.block_confirmed (transaction_a, hash_a);
 }
 
 nano::uint128_t nano::node::delta () const
@@ -1272,32 +1272,23 @@ void nano::node::process_confirmed_data (nano::transaction const & transaction_a
 
 void nano::node::process_confirmed (nano::election_status const & status_a, std::shared_ptr<nano::election> const & election_a, uint8_t iteration_a)
 {
-	if (status_a.type == nano::election_status_type::active_confirmed_quorum)
+	auto block_a (status_a.winner);
+	auto hash (block_a->hash ());
+	if (ledger.block_exists (block_a->type (), hash))
 	{
-		auto block_a (status_a.winner);
-		auto hash (block_a->hash ());
-		if (ledger.block_exists (block_a->type (), hash))
-		{
-			// Pausing to prevent this block being processed before adding to election winner details.
-			confirmation_height_processor.pause ();
-			confirmation_height_processor.add (hash);
+		confirmation_height_processor.add (hash);
+	}
+	// Limit to 0.5 * 20 = 10 seconds (more than max block_processor::process_batch finish time)
+	else if (iteration_a < 20)
+	{
+		iteration_a++;
+		std::weak_ptr<nano::node> node_w (shared ());
+		alarm.add (std::chrono::steady_clock::now () + network_params.node.process_confirmed_interval, [node_w, status_a, iteration_a, election_a]() {
+			if (auto node_l = node_w.lock ())
 			{
-				active.add_election_winner_details (hash, election_a);
+				node_l->process_confirmed (status_a, election_a, iteration_a);
 			}
-			confirmation_height_processor.unpause ();
-		}
-		// Limit to 0.5 * 20 = 10 seconds (more than max block_processor::process_batch finish time)
-		else if (iteration_a < 20)
-		{
-			iteration_a++;
-			std::weak_ptr<nano::node> node_w (shared ());
-			alarm.add (std::chrono::steady_clock::now () + network_params.node.process_confirmed_interval, [node_w, status_a, iteration_a, election_a]() {
-				if (auto node_l = node_w.lock ())
-				{
-					node_l->process_confirmed (status_a, election_a, iteration_a);
-				}
-			});
-		}
+		});
 	}
 }
 

nano/node/vote_processor.cpp

@@ -165,7 +165,6 @@ void nano::vote_processor::verify_votes (decltype (votes) const & votes_a)
 	}
 }
 
-// node.active.mutex lock required
 nano::vote_code nano::vote_processor::vote_blocking (std::shared_ptr<nano::vote> vote_a, std::shared_ptr<nano::transport::channel> channel_a, bool validated)
 {
 	auto result (nano::vote_code::invalid);