Allow restarting elections with higher work (#2715)

* Allow restarting elections with higher work

* Use debug_assert

* Simpler return variable (Wes review)

* Add missing transition_active, move insertion in recently_confirmed to election cleanup

* Test condition fix

nano/core_test/active_transactions.cpp

@@ -213,6 +213,7 @@ TEST (active_transactions, keep_local)
 	{
 		ASSERT_NO_ERROR (system.poll ());
 	}
+	ASSERT_EQ (0, node.active.recently_dropped.size ());
 	while (!node.active.empty ())
 	{
 		nano::lock_guard<std::mutex> active_guard (node.active.mutex);
@@ -234,6 +235,7 @@ TEST (active_transactions, keep_local)
 	{
 		ASSERT_NO_ERROR (system.poll ());
 	}
+	ASSERT_EQ (1, node.active.recently_dropped.size ());
 }
 
 TEST (active_transactions, prioritize_chains)
@@ -1025,3 +1027,50 @@ TEST (active_transactions, confirm_new)
 		ASSERT_NO_ERROR (system.poll ());
 	}
 }
+
+TEST (active_transactions, restart_dropped)
+{
+	nano::system system;
+	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);
+	nano::genesis genesis;
+	auto send (std::make_shared<nano::state_block> (nano::test_genesis_key.pub, genesis.hash (), nano::test_genesis_key.pub, nano::genesis_amount - nano::xrb_ratio, nano::test_genesis_key.pub, nano::test_genesis_key.prv, nano::test_genesis_key.pub, *system.work.generate (genesis.hash ())));
+	// Process only in ledger and simulate dropping the election
+	ASSERT_EQ (nano::process_result::progress, node.process (*send).code);
+	node.active.recently_dropped.add (send->qualified_root ());
+	// Generate higher difficulty work
+	ASSERT_TRUE (node.work_generate_blocking (*send, send->difficulty () + 1).is_initialized ());
+	// Process the same block with updated work
+	ASSERT_EQ (0, node.active.size ());
+	node.process_active (send);
+	node.block_processor.flush ();
+	ASSERT_EQ (1, node.active.size ());
+	auto ledger_block (node.store.block_get (node.store.tx_begin_read (), send->hash ()));
+	ASSERT_NE (nullptr, ledger_block);
+	// Exact same block, including work value must have been re-written
+	ASSERT_EQ (*send, *ledger_block);
+	// Removed from the dropped elections cache
+	ASSERT_EQ (std::chrono::steady_clock::time_point{}, node.active.recently_dropped.find (send->qualified_root ()));
+	// Drop election
+	node.active.erase (*send);
+	ASSERT_EQ (0, node.active.size ());
+	// Try to restart election with the same difficulty
+	node.process_active (send);
+	node.block_processor.flush ();
+	ASSERT_EQ (0, node.active.size ());
+	// Verify the block was not updated in the ledger
+	ASSERT_EQ (*node.store.block_get (node.store.tx_begin_read (), send->hash ()), *send);
+	// Generate even higher difficulty work
+	ASSERT_TRUE (node.work_generate_blocking (*send, send->difficulty () + 1).is_initialized ());
+	// Add voting
+	system.wallet (0)->insert_adhoc (nano::test_genesis_key.prv);
+	// Process the same block with updated work
+	ASSERT_EQ (0, node.active.size ());
+	node.process_active (send);
+	node.block_processor.flush ();
+	ASSERT_EQ (1, node.active.size ());
+	ASSERT_EQ (1, node.ledger.cache.cemented_count);
+	// Wait for the election to complete
+	ASSERT_TIMELY (5s, node.ledger.cache.cemented_count == 2);
+}

nano/node/active_transactions.cpp

@@ -93,10 +93,10 @@ void nano::active_transactions::confirm_prioritized_frontiers (nano::transaction
 						if (info.block_count > confirmation_height_info.height)
 						{
 							auto block (this->node.store.block_get (transaction_a, info.head));
-							auto election_insert_result = this->insert (block);
-							if (election_insert_result.inserted)
+							auto insert_result = this->insert (block);
+							if (insert_result.inserted)
 							{
-								election_insert_result.election->transition_active ();
+								insert_result.election->transition_active ();
 								++elections_count;
 							}
 						}
@@ -635,18 +635,21 @@ std::shared_ptr<nano::election> nano::active_transactions::election (nano::quali
 	return result;
 }
 
-void nano::active_transactions::update_difficulty (nano::block const & block_a)
+bool nano::active_transactions::update_difficulty (nano::block const & block_a)
 {
-	nano::unique_lock<std::mutex> lock (mutex);
+	nano::lock_guard<std::mutex> guard (mutex);
 	auto existing_election (roots.get<tag_root> ().find (block_a.qualified_root ()));
-	if (existing_election != roots.get<tag_root> ().end ())
+	auto found = existing_election != roots.get<tag_root> ().end ();
+	if (found)
 	{
 		update_difficulty_impl (existing_election, block_a);
 	}
+	return !found;
 }
 
 void nano::active_transactions::update_difficulty_impl (nano::active_transactions::roots_iterator const & root_it_a, nano::block const & block_a)
 {
+	debug_assert (!mutex.try_lock ());
 	double multiplier (normalized_multiplier (block_a, root_it_a));
 	if (multiplier > root_it_a->multiplier)
 	{
@@ -661,6 +664,39 @@ void nano::active_transactions::update_difficulty_impl (nano::active_transaction
 	}
 }
 
+void nano::active_transactions::restart (std::shared_ptr<nano::block> const & block_a, nano::write_transaction const & transaction_a)
+{
+	// Only guaranteed to restart the election if the new block is received within 2 minutes of its election being dropped
+	constexpr std::chrono::minutes recently_dropped_cutoff{ 2 };
+	if (recently_dropped.find (block_a->qualified_root ()) > std::chrono::steady_clock::now () - recently_dropped_cutoff)
+	{
+		auto hash (block_a->hash ());
+		auto ledger_block (node.store.block_get (transaction_a, hash));
+		if (ledger_block != nullptr && ledger_block->block_work () != block_a->block_work () && !node.block_confirmed_or_being_confirmed (transaction_a, hash))
+		{
+			if (block_a->difficulty () > ledger_block->difficulty ())
+			{
+				// Re-writing the block is necessary to avoid the same work being received later to force restarting the election
+				// The existing block is re-written, not the arriving block, as that one might not have gone through a full signature check
+				ledger_block->block_work_set (block_a->block_work ());
+
+				auto block_count = node.ledger.cache.block_count.load ();
+				node.store.block_put (transaction_a, hash, *ledger_block);
+				debug_assert (node.ledger.cache.block_count.load () == block_count);
+
+				// Restart election for the upgraded block, previously dropped from elections
+				auto previous_balance = node.ledger.balance (transaction_a, ledger_block->previous ());
+				auto insert_result = insert (ledger_block, previous_balance);
+				if (insert_result.inserted)
+				{
+					insert_result.election->transition_active ();
+					recently_dropped.erase (ledger_block->qualified_root ());
+				}
+			}
+		}
+	}
+}
+
 double nano::active_transactions::normalized_multiplier (nano::block const & block_a, boost::optional<nano::active_transactions::roots_iterator> const & root_it_a) const
 {
 	debug_assert (!mutex.try_lock ());
@@ -903,13 +939,14 @@ void nano::active_transactions::add_recently_confirmed (nano::qualified_root con
 
 void nano::active_transactions::erase (nano::block const & block_a)
 {
-	nano::lock_guard<std::mutex> lock (mutex);
+	nano::unique_lock<std::mutex> lock (mutex);
 	auto root_it (roots.get<tag_root> ().find (block_a.qualified_root ()));
 	if (root_it != roots.get<tag_root> ().end ())
 	{
 		root_it->election->cleanup ();
 		root_it->election->adjust_dependent_difficulty ();
 		roots.get<tag_root> ().erase (root_it);
+		lock.unlock ();
 		node.logger.try_log (boost::str (boost::format ("Election erased for block block %1% root %2%") % block_a.hash ().to_string () % block_a.root ().to_string ()));
 	}
 }
@@ -1146,3 +1183,41 @@ std::unique_ptr<nano::container_info_component> nano::collect_container_info (ac
 	composite->add_component (collect_container_info (active_transactions.generator, "generator"));
 	return composite;
 }
+
+void nano::dropped_elections::add (nano::qualified_root const & root_a)
+{
+	nano::lock_guard<std::mutex> guard (mutex);
+	auto & items_by_sequence = items.get<tag_sequence> ();
+	items_by_sequence.emplace_back (nano::election_timepoint{ std::chrono::steady_clock::now (), root_a });
+	if (items.size () > capacity)
+	{
+		items_by_sequence.pop_front ();
+	}
+}
+
+void nano::dropped_elections::erase (nano::qualified_root const & root_a)
+{
+	nano::lock_guard<std::mutex> guard (mutex);
+	items.get<tag_root> ().erase (root_a);
+}
+
+std::chrono::steady_clock::time_point nano::dropped_elections::find (nano::qualified_root const & root_a) const
+{
+	nano::lock_guard<std::mutex> guard (mutex);
+	auto & items_by_root = items.get<tag_root> ();
+	auto existing (items_by_root.find (root_a));
+	if (existing != items_by_root.end ())
+	{
+		return existing->time;
+	}
+	else
+	{
+		return std::chrono::steady_clock::time_point{};
+	}
+}
+
+size_t nano::dropped_elections::size () const
+{
+	nano::lock_guard<std::mutex> guard (mutex);
+	return items.size ();
+}

nano/node/active_transactions.hpp

@@ -57,6 +57,31 @@ public:
 	bool confirmed{ false }; // Did item reach votes quorum? (minimum config value)
 };
 
+class dropped_elections final
+{
+public:
+	void add (nano::qualified_root const &);
+	void erase (nano::qualified_root const &);
+	std::chrono::steady_clock::time_point find (nano::qualified_root const &) const;
+	size_t size () const;
+
+	static size_t constexpr capacity{ 16 * 1024 };
+
+	// clang-format off
+	class tag_sequence {};
+	class tag_root {};
+	using ordered_dropped = boost::multi_index_container<nano::election_timepoint,
+	mi::indexed_by<
+		mi::sequenced<mi::tag<tag_sequence>>,
+		mi::hashed_unique<mi::tag<tag_root>,
+			mi::member<nano::election_timepoint, decltype(nano::election_timepoint::root), &nano::election_timepoint::root>>>>;
+	// clang-format on
+
+private:
+	ordered_dropped items;
+	mutable std::mutex mutex;
+};
+
 class election_insertion_result final
 {
 public:
@@ -117,7 +142,9 @@ public:
 	bool active (nano::block const &);
 	bool active (nano::qualified_root const &);
 	std::shared_ptr<nano::election> election (nano::qualified_root const &) const;
-	void update_difficulty (nano::block const &);
+	// Returns true if this block was not active
+	bool update_difficulty (nano::block const &);
+	void restart (std::shared_ptr<nano::block> const &, nano::write_transaction const &);
 	double normalized_multiplier (nano::block const &, boost::optional<roots_iterator> const & = boost::none) const;
 	void add_adjust_difficulty (nano::block_hash const &);
 	void update_adjusted_multiplier ();
@@ -139,6 +166,7 @@ public:
 	std::unordered_map<nano::block_hash, std::shared_ptr<nano::election>> blocks;
 	std::deque<nano::election_status> list_recently_cemented ();
 	std::deque<nano::election_status> recently_cemented;
+	dropped_elections recently_dropped;
 
 	void add_recently_cemented (nano::election_status const &);
 	void add_recently_confirmed (nano::qualified_root const &, nano::block_hash const &);

nano/node/blockprocessor.cpp

@@ -368,7 +368,10 @@ nano::process_return nano::block_processor::process_one (nano::write_transaction
 				node.logger.try_log (boost::str (boost::format ("Old for: %1%") % hash.to_string ()));
 			}
 			queue_unchecked (transaction_a, hash);
-			node.active.update_difficulty (*info_a.block);
+			if (node.active.update_difficulty (*info_a.block))
+			{
+				node.active.restart (info_a.block, transaction_a);
+			}
 			node.stats.inc (nano::stat::type::ledger, nano::stat::detail::old);
 			break;
 		}

nano/node/election.cpp

@@ -550,6 +550,7 @@ void nano::election::adjust_dependent_difficulty ()
 void nano::election::cleanup ()
 {
 	bool unconfirmed (!confirmed ());
+	auto winner_root (status.winner->qualified_root ());
 	auto winner_hash (status.winner->hash ());
 	for (auto const & block : blocks)
 	{
@@ -566,6 +567,8 @@ void nano::election::cleanup ()
 	}
 	if (unconfirmed)
 	{
+		node.active.recently_dropped.add (winner_root);
+
 		// Clear network filter in another thread
 		node.worker.push_task ([node_l = node.shared (), blocks_l = std::move (blocks)]() {
 			for (auto const & block : blocks_l)