[v17] Replay votes in response to a confirm_req for an active block

rai/core_test/node.cpp

@@ -363,7 +363,10 @@ TEST (node, unlock_search)
 		ASSERT_NO_ERROR (system.poll ());
 	}
 	system.wallet (0)->insert_adhoc (key2.prv);
-	system.wallet (0)->store.password.value_set (rai::keypair ().prv);
+	{
+		std::lock_guard<std::recursive_mutex> lock (system.wallet (0)->store.mutex);
+		system.wallet (0)->store.password.value_set (rai::keypair ().prv);
+	}
 	auto node (system.nodes[0]);
 	{
 		auto transaction (system.wallet (0)->wallets.tx_begin (true));
@@ -1915,3 +1918,48 @@ TEST (node, block_processor_reject_state)
 	node.block_processor.flush ();
 	ASSERT_TRUE (node.ledger.block_exists (send2->hash ()));
 }
+
+TEST (node, confirm_req_active)
+{
+	rai::system system (24000, 2);
+	rai::keypair key1;
+	system.wallet (0)->insert_adhoc (key1.prv);
+	rai::genesis genesis;
+	auto amount (rai::genesis_amount / 100);
+	std::shared_ptr<rai::block> send1 (std::make_shared<rai::send_block> (genesis.hash (), key1.pub, rai::genesis_amount - amount, rai::test_genesis_key.prv, rai::test_genesis_key.pub, system.work.generate (genesis.hash ())));
+	system.nodes[0]->process_active (send1);
+	system.deadline_set (5s);
+	while (!system.nodes[0]->block (send1->hash ()))
+	{
+		ASSERT_NO_ERROR (system.poll ());
+	}
+	// Force a receive to assign the stake
+	system.wallet (0)->receive_async (send1, key1.pub, amount, [](std::shared_ptr<rai::block>) {});
+	while (!system.nodes[1]->active.active (*send1))
+	{
+		ASSERT_NO_ERROR (system.poll ());
+	}
+	while (system.nodes[1]->stats.count (rai::stat::type::message, rai::stat::detail::confirm_ack, rai::stat::dir::in) == 0)
+	{
+		ASSERT_NO_ERROR (system.poll ());
+	}
+	for (size_t i = 0; i < 20; ++i)
+	{
+		system.nodes[1]->network.broadcast_confirm_req (send1);
+	}
+	while (system.nodes[0]->stats.count (rai::stat::type::message, rai::stat::detail::confirm_req, rai::stat::dir::in) < 20)
+	{
+		ASSERT_NO_ERROR (system.poll ());
+	}
+	while (system.nodes[1]->stats.count (rai::stat::type::message, rai::stat::detail::confirm_ack, rai::stat::dir::in) < 20)
+	{
+		ASSERT_NO_ERROR (system.poll ());
+	}
+	{
+		auto transaction (system.nodes[0]->store.tx_begin_read ());
+		std::lock_guard<std::mutex> guard (boost::polymorphic_downcast<rai::mdb_store *> (system.nodes[0]->store_impl.get ())->cache_mutex);
+		auto vote (system.nodes[0]->store.vote_current (transaction, key1.pub));
+		ASSERT_NE (vote, nullptr);
+		ASSERT_LT (vote->sequence, 20);
+	}
+}

rai/node/node.cpp

@@ -540,12 +540,41 @@ class network_message_visitor : public rai::message_visitor
 		// Don't load nodes with disabled voting
 		if (node.config.enable_voting)
 		{
-			auto transaction (node.store.tx_begin_read ());
-			auto successor (node.ledger.successor (transaction, message_a.block->root ()));
-			if (successor != nullptr)
+			std::unique_lock<std::mutex> active_lock (node.active.mutex);
+			auto active_it (node.active.roots.get<0> ().find (message_a.block->root ()));
+			if (active_it != node.active.roots.get<0> ().end ())
 			{
-				auto same_block (successor->hash () == message_a.block->hash ());
-				confirm_block (transaction, node, sender, std::move (successor), !same_block);
+				// Replay votes in response to a confirm_req for an active block
+				for (auto & it : active_it->election->our_last_votes)
+				{
+					rai::confirm_ack confirm (it.second);
+					std::shared_ptr<std::vector<uint8_t>> vote_bytes (new std::vector<uint8_t>);
+					{
+						rai::vectorstream stream (*vote_bytes);
+						confirm.serialize (stream);
+					}
+					node.network.confirm_send (confirm, vote_bytes, sender);
+				}
+				rai::publish publish (active_it->election->status.winner);
+				active_lock.unlock ();
+				std::shared_ptr<std::vector<uint8_t>> publish_bytes (new std::vector<uint8_t>);
+				{
+					rai::vectorstream stream (*publish_bytes);
+					publish.serialize (stream);
+				}
+				node.network.republish (publish.block->hash (), publish_bytes, sender);
+			}
+			else
+			{
+				active_lock.unlock ();
+				// Generating new vote
+				auto transaction (node.store.tx_begin_read ());
+				auto successor (node.ledger.successor (transaction, message_a.block->root ()));
+				if (successor != nullptr)
+				{
+					auto same_block (successor->hash () == message_a.block->hash ());
+					confirm_block (transaction, node, sender, std::move (successor), !same_block);
+				}
 			}
 		}
 	}
@@ -2717,6 +2746,7 @@ void rai::election::compute_rep_votes (rai::transaction const & transaction_a)
 	{
 		node.wallets.foreach_representative (transaction_a, [this, &transaction_a](rai::public_key const & pub_a, rai::raw_key const & prv_a) {
 			auto vote (this->node.store.vote_generate (transaction_a, pub_a, prv_a, status.winner));
+			this->our_last_votes[pub_a] = vote;
 			this->node.vote_processor.vote (vote, this->node.network.endpoint ());
 		});
 	}
@@ -3017,6 +3047,12 @@ void rai::active_transactions::announce_votes (std::unique_lock<std::mutex> & lo
 							}
 						}
 					}
+					/* Regenerate votes for long unconfirmed blocks
+					Useful for tests & large representatives */
+					if (node.config.enable_voting && election_l->our_last_votes.empty ())
+					{
+						node.block_processor.generator.add (election_l->status.winner->hash ());
+					}
 				}
 			}
 			if (i->election->announcements < announcement_long || i->election->announcements % announcement_long == 1)

rai/node/node.hpp

@@ -65,6 +65,7 @@ class election : public std::enable_shared_from_this<rai::election>
 	void stop ();
 	rai::node & node;
 	std::unordered_map<rai::account, rai::vote_info> last_votes;
+	std::unordered_map<rai::account, std::shared_ptr<rai::vote>> our_last_votes;
 	std::unordered_map<rai::block_hash, std::shared_ptr<rai::block>> blocks;
 	rai::block_hash root;
 	rai::election_status status;
@@ -384,6 +385,7 @@ class block_processor
 	bool have_blocks ();
 	void process_blocks ();
 	rai::process_return process_receive_one (rai::transaction const &, std::shared_ptr<rai::block>, std::chrono::steady_clock::time_point = std::chrono::steady_clock::now (), bool = false);
+	rai::vote_generator generator;
 
 private:
 	void queue_unchecked (rai::transaction const &, rai::block_hash const &);
@@ -398,7 +400,6 @@ class block_processor
 	std::deque<std::shared_ptr<rai::block>> forced;
 	std::condition_variable condition;
 	rai::node & node;
-	rai::vote_generator generator;
 	std::mutex mutex;
 };
 class node : public std::enable_shared_from_this<rai::node>

rai/node/voting.cpp

@@ -49,6 +49,16 @@ void rai::vote_generator::send (std::unique_lock<std::mutex> & lock_a)
 		auto transaction (node.store.tx_begin_read ());
 		node.wallets.foreach_representative (transaction, [this, &hashes_l, &transaction](rai::public_key const & pub_a, rai::raw_key const & prv_a) {
 			auto vote (this->node.store.vote_generate (transaction, pub_a, prv_a, hashes_l));
+			std::unique_lock<std::mutex> lock (this->node.active.mutex);
+			for (auto & hash : hashes_l)
+			{
+				auto existing (this->node.active.successors.find (hash));
+				if (existing != this->node.active.successors.end ())
+				{
+					existing->second->our_last_votes[pub_a] = vote;
+				}
+			}
+			lock.unlock ();
 			this->node.vote_processor.vote (vote, this->node.network.endpoint ());
 		});
 	}

rai/node/wallet.cpp

@@ -1385,6 +1385,7 @@ void rai::wallets::foreach_representative (rai::transaction const & transaction_
 	for (auto i (items.begin ()), n (items.end ()); i != n; ++i)
 	{
 		auto & wallet (*i->second);
+		std::lock_guard<std::recursive_mutex> lock (wallet.store.mutex);
 		for (auto j (wallet.store.begin (transaction_a)), m (wallet.store.end ()); j != m; ++j)
 		{
 			rai::account account (j->first);