Prevent getting stuck in block processor flush (#2675)

* Prevent getting stuck in block processor flush

I noticed `node.block_processor_reject_state` was often freezing on windows, this is due to a the verification callback being called and notifying the block processor before transitioning into an inactive state, so the `block_processor::flush` ends up waiting for the condition forever. I've added a second callback to solve this.

* Lock before notifying to prevent a race with condition.wait; check if flushing first

* Change the test to launch async and wait for future, to prevent freezing in the future but still fail

* Comment formatting

* Add comment on why lock before notifying (all Wesley review)

nano/core_test/node.cpp

@@ -3193,6 +3193,7 @@ TEST (node, block_processor_signatures)
 
 /*
  *  State blocks go through a different signature path, ensure invalidly signed state blocks are rejected
+ *  This test can freeze if the wake conditions in block_processor::flush are off, for that reason this is done async here
  */
 TEST (node, block_processor_reject_state)
 {
@@ -3204,12 +3205,14 @@ TEST (node, block_processor_reject_state)
 	send1->signature.bytes[0] ^= 1;
 	ASSERT_FALSE (node.ledger.block_exists (send1->hash ()));
 	node.process_active (send1);
-	node.block_processor.flush ();
+	auto flushed = std::async (std::launch::async, [&node] { node.block_processor.flush (); });
+	ASSERT_NE (std::future_status::timeout, flushed.wait_for (5s));
 	ASSERT_FALSE (node.ledger.block_exists (send1->hash ()));
 	auto send2 (std::make_shared<nano::state_block> (nano::test_genesis_key.pub, genesis.hash (), nano::test_genesis_key.pub, nano::genesis_amount - 2 * nano::Gxrb_ratio, nano::test_genesis_key.pub, nano::test_genesis_key.prv, nano::test_genesis_key.pub, 0));
 	node.work_generate_blocking (*send2);
 	node.process_active (send2);
-	node.block_processor.flush ();
+	auto flushed2 = std::async (std::launch::async, [&node] { node.block_processor.flush (); });
+	ASSERT_NE (std::future_status::timeout, flushed2.wait_for (5s));
 	ASSERT_TRUE (node.ledger.block_exists (send2->hash ()));
 }
 

nano/node/blockprocessor.cpp

@@ -21,6 +21,16 @@ state_block_signature_verification (node.checker, node.ledger.network_params.led
 	state_block_signature_verification.blocks_verified_callback = [this](std::deque<nano::unchecked_info> & items, std::vector<int> const & verifications, std::vector<nano::block_hash> const & hashes, std::vector<nano::signature> const & blocks_signatures) {
 		this->process_verified_state_blocks (items, verifications, hashes, blocks_signatures);
 	};
+	state_block_signature_verification.transition_inactive_callback = [this]() {
+		if (this->flushing)
+		{
+			{
+				// Prevent a race with condition.wait in block_processor::flush
+				nano::lock_guard<std::mutex> guard (this->mutex);
+			}
+			this->condition.notify_all ();
+		}
+	};
 }
 
 nano::block_processor::~block_processor ()

nano/node/state_block_signature_verification.cpp

@@ -56,6 +56,9 @@ void nano::state_block_signature_verification::run (uint64_t state_block_signatu
 				lk.lock ();
 			}
 			active = false;
+			lk.unlock ();
+			transition_inactive_callback ();
+			lk.lock ();
 		}
 		else
 		{

nano/node/state_block_signature_verification.hpp

@@ -25,6 +25,7 @@ class state_block_signature_verification
 	bool is_active ();
 
 	std::function<void(std::deque<nano::unchecked_info> &, std::vector<int> const &, std::vector<nano::block_hash> const &, std::vector<nano::signature> const &)> blocks_verified_callback;
+	std::function<void()> transition_inactive_callback;
 
 private:
 	nano::signature_checker & signature_checker;