Wallets search pending deadlock (#2983)

* Add disable_search_pending flag and a deadlocking test on wallet search_pending[_all]

* Unlock wallets mutex before wallet::search_pending, fixing the deadlock

* Using wallet::open

* receive_async directly in wallet::search_pending

This avoids a dependency inversion as node::receive_confirmed searches all wallets. wallet::search_pending we already know the wallet responsible for receiving that block.

* Use wallets.open() in json_handler::wallet_impl

nano/core_test/wallet.cpp

@@ -1458,7 +1458,9 @@ TEST (wallet, search_pending)
 	nano::node_config config (nano::get_available_port (), system.logging);
 	config.enable_voting = false;
 	config.frontiers_confirmation = nano::frontiers_confirmation_mode::disabled;
-	auto & node (*system.add_node ());
+	nano::node_flags flags;
+	flags.disable_search_pending = true;
+	auto & node (*system.add_node (config, flags));
 	auto & wallet (*system.wallet (0));
 
 	wallet.insert_adhoc (nano::dev_genesis_key.prv);

nano/core_test/wallets.cpp

@@ -164,3 +164,76 @@ TEST (wallets, exists)
 		ASSERT_TRUE (node.wallets.exists (transaction, key2.pub));
 	}
 }
+
+TEST (wallets, search_pending)
+{
+	for (auto search_all : { false, true })
+	{
+		nano::system system;
+		nano::node_config config (nano::get_available_port (), system.logging);
+		config.enable_voting = false;
+		config.frontiers_confirmation = nano::frontiers_confirmation_mode::disabled;
+		nano::node_flags flags;
+		flags.disable_search_pending = true;
+		auto & node (*system.add_node (config, flags));
+
+		auto wallets = node.wallets.get_wallets ();
+		ASSERT_EQ (1, wallets.size ());
+		auto wallet_id = wallets.begin ()->first;
+		auto wallet = wallets.begin ()->second;
+
+		wallet->insert_adhoc (nano::dev_genesis_key.prv);
+		nano::block_builder builder;
+		auto send = builder.state ()
+		            .account (nano::genesis_account)
+		            .previous (nano::genesis_hash)
+		            .representative (nano::genesis_account)
+		            .balance (nano::genesis_amount - node.config.receive_minimum.number ())
+		            .link (nano::genesis_account)
+		            .sign (nano::dev_genesis_key.prv, nano::dev_genesis_key.pub)
+		            .work (*system.work.generate (nano::genesis_hash))
+		            .build ();
+		ASSERT_EQ (nano::process_result::progress, node.process (*send).code);
+
+		// Pending search should start an election
+		ASSERT_TRUE (node.active.empty ());
+		if (search_all)
+		{
+			node.wallets.search_pending_all ();
+		}
+		else
+		{
+			node.wallets.search_pending (wallet_id);
+		}
+		auto election = node.active.election (send->qualified_root ());
+		ASSERT_NE (nullptr, election);
+
+		// Erase the key so the confirmation does not trigger an automatic receive
+		wallet->store.erase (node.wallets.tx_begin_write (), nano::genesis_account);
+
+		// Now confirm the election
+		election->force_confirm ();
+
+		ASSERT_TIMELY (5s, node.block_confirmed (send->hash ()) && node.active.empty ());
+
+		// Re-insert the key
+		wallet->insert_adhoc (nano::dev_genesis_key.prv);
+
+		// Pending search should create the receive block
+		ASSERT_EQ (2, node.ledger.cache.block_count);
+		if (search_all)
+		{
+			node.wallets.search_pending_all ();
+		}
+		else
+		{
+			node.wallets.search_pending (wallet_id);
+		}
+		ASSERT_TIMELY (3s, node.balance (nano::genesis_account) == nano::genesis_amount);
+		auto receive_hash = node.ledger.latest (node.store.tx_begin_read (), nano::genesis_account);
+		auto receive = node.block (receive_hash);
+		ASSERT_NE (nullptr, receive);
+		ASSERT_EQ (receive->sideband ().height, 3);
+		ASSERT_EQ (send->hash (), receive->link ().as_block_hash ());
+	}
+}

nano/node/json_handler.cpp

@@ -177,10 +177,9 @@ std::shared_ptr<nano::wallet> nano::json_handler::wallet_impl ()
 		nano::wallet_id wallet;
 		if (!wallet.decode_hex (wallet_text))
 		{
-			auto existing (node.wallets.items.find (wallet));
+			if (auto existing = node.wallets.open (wallet); existing != nullptr)
-			if (existing != node.wallets.items.end ())
 			{
-				return existing->second;
+				return existing;
 			}
 			else
 			{

nano/node/node.cpp

@@ -669,7 +669,10 @@ void nano::node::start ()
 	{
 		backup_wallet ();
 	}
+	if (!flags.disable_search_pending)
+	{
 		search_pending ();
+	}
 	if (!flags.disable_wallet_bootstrap)
 	{
 		// Delay to start wallet lazy bootstrap

nano/node/nodeconfig.hpp

@@ -141,6 +141,7 @@ public:
 	bool allow_bootstrap_peers_duplicates{ false };
 	bool disable_max_peers_per_ip{ false }; // For testing only
 	bool force_use_write_database_queue{ false }; // For testing only. RocksDB does not use the database queue, but some tests rely on it being used.
+	bool disable_search_pending{ false }; // For testing only
 	bool enable_pruning{ false };
 	bool fast_bootstrap{ false };
 	bool read_only{ false };

nano/node/wallet.cpp

@@ -1193,8 +1193,9 @@ bool nano::wallet::search_pending ()
 						auto block (wallets.node.store.block_get (block_transaction, hash));
 						if (wallets.node.ledger.block_confirmed (block_transaction, hash))
 						{
+							auto representative = store.representative (wallet_transaction);
 							// Receive confirmed block
-							wallets.node.receive_confirmed (wallet_transaction, block_transaction, block, hash);
+							receive_async (block, representative, amount, [](std::shared_ptr<nano::block>) {});
 						}
 						else if (!wallets.node.confirmation_height_processor.is_processing_block (hash))
 						{
@@ -1552,13 +1553,9 @@ std::shared_ptr<nano::wallet> nano::wallets::create (nano::wallet_id const & id_
 
 bool nano::wallets::search_pending (nano::wallet_id const & wallet_a)
 {
-	nano::lock_guard<std::mutex> lock (mutex);
 	auto result (false);
-	auto existing (items.find (wallet_a));
+	if (auto wallet = open (wallet_a); wallet != nullptr)
-	result = existing == items.end ();
-	if (!result)
 	{
-		auto wallet (existing->second);
 		result = wallet->search_pending ();
 	}
 	return result;
@@ -1566,10 +1563,9 @@ bool nano::wallets::search_pending (nano::wallet_id const & wallet_a)
 
 void nano::wallets::search_pending_all ()
 {
-	nano::lock_guard<std::mutex> lock (mutex);
+	for (auto const & [id, wallet] : get_wallets ())
-	for (auto i : items)
 	{
-		i.second->search_pending ();
+		wallet->search_pending ();
 	}
 }