Move node::receive_confirmed to wallets::receive_confirmed (#4557)

This is inherently a wallet operation so it's more appropriate to be on the wallets class.
This also breaks a direct dependency/cycle from block processing to the wallet operations.
This also breaks a lock order inversion where a ledger transaction mutex is held while acquiring the wallet mutexes while most operations do it in the reverse order.

nano/node/node.cpp

@@ -468,8 +468,9 @@ nano::node::node (std::shared_ptr<boost::asio::io_context> io_ctx_a, std::filesy
 		confirming_set.cemented_observers.add ([this] (auto const & block) {
 			if (block->is_send ())
 			{
-				auto transaction = store.tx_begin_read ();
+				workers.push_task ([this, hash = block->hash (), destination = block->destination ()] () {
-				receive_confirmed (transaction, block->hash (), block->destination ());
+					wallets.receive_confirmed (hash, destination);
+				});
 			}
 		});
 	}
@@ -1217,40 +1218,6 @@ void nano::node::ongoing_online_weight_calculation ()
 	ongoing_online_weight_calculation_queue ();
 }
 
-void nano::node::receive_confirmed (store::transaction const & block_transaction_a, nano::block_hash const & hash_a, nano::account const & destination_a)
-{
-	nano::unique_lock<nano::mutex> lk{ wallets.mutex };
-	auto wallets_l = wallets.get_wallets ();
-	auto wallet_transaction = wallets.tx_begin_read ();
-	lk.unlock ();
-	for ([[maybe_unused]] auto const & [id, wallet] : wallets_l)
-	{
-		if (wallet->store.exists (wallet_transaction, destination_a))
-		{
-			nano::account representative;
-			representative = wallet->store.representative (wallet_transaction);
-			auto pending = ledger.pending_info (block_transaction_a, nano::pending_key (destination_a, hash_a));
-			if (pending)
-			{
-				auto amount (pending->amount.number ());
-				wallet->receive_async (hash_a, representative, amount, destination_a, [] (std::shared_ptr<nano::block> const &) {});
-			}
-			else
-			{
-				if (!ledger.block_or_pruned_exists (block_transaction_a, hash_a))
-				{
-					logger.warn (nano::log::type::node, "Confirmed block is missing: {}", hash_a.to_string ());
-					debug_assert (false, "Confirmed block is missing");
-				}
-				else
-				{
-					logger.warn (nano::log::type::node, "Block has already been received: {}", hash_a.to_string ());
-				}
-			}
-		}
-	}
-}
-
 void nano::node::process_confirmed (nano::election_status const & status_a, uint64_t iteration_a)
 {
 	auto hash (status_a.winner->hash ());

nano/node/node.hpp

@@ -87,7 +87,6 @@ public:
 	void stop ();
 	std::shared_ptr<nano::node> shared ();
 	int store_version ();
-	void receive_confirmed (store::transaction const & block_transaction_a, nano::block_hash const & hash_a, nano::account const & destination_a);
 	void process_confirmed (nano::election_status const &, uint64_t = 0);
 	void process_active (std::shared_ptr<nano::block> const &);
 	std::optional<nano::block_status> process_local (std::shared_ptr<nano::block> const &);

nano/node/wallet.cpp

@@ -1707,6 +1707,40 @@ void nano::wallets::ongoing_compute_reps ()
 	});
 }
 
+void nano::wallets::receive_confirmed (nano::block_hash const & hash_a, nano::account const & destination_a)
+{
+	nano::unique_lock<nano::mutex> lk{ mutex };
+	auto wallets_l = get_wallets ();
+	auto wallet_transaction = tx_begin_read ();
+	lk.unlock ();
+	for ([[maybe_unused]] auto const & [id, wallet] : wallets_l)
+	{
+		if (wallet->store.exists (wallet_transaction, destination_a))
+		{
+			nano::account representative;
+			representative = wallet->store.representative (wallet_transaction);
+			auto pending = node.ledger.pending_info (node.ledger.store.tx_begin_read (), nano::pending_key (destination_a, hash_a));
+			if (pending)
+			{
+				auto amount (pending->amount.number ());
+				wallet->receive_async (hash_a, representative, amount, destination_a, [] (std::shared_ptr<nano::block> const &) {});
+			}
+			else
+			{
+				if (!node.ledger.block_or_pruned_exists (node.ledger.store.tx_begin_read (), hash_a))
+				{
+					node.logger.warn (nano::log::type::wallet, "Confirmed block is missing: {}", hash_a.to_string ());
+					debug_assert (false, "Confirmed block is missing");
+				}
+				else
+				{
+					node.logger.warn (nano::log::type::wallet, "Block has already been received: {}", hash_a.to_string ());
+				}
+			}
+		}
+	}
+}
+
 std::unordered_map<nano::wallet_id, std::shared_ptr<nano::wallet>> nano::wallets::get_wallets ()
 {
 	debug_assert (!mutex.try_lock ());

nano/node/wallet.hpp

@@ -220,6 +220,7 @@ public:
 	bool check_rep (nano::account const &, nano::uint128_t const &, bool const = true);
 	void compute_reps ();
 	void ongoing_compute_reps ();
+	void receive_confirmed (nano::block_hash const & hash_a, nano::account const & destination_a);
 	std::unordered_map<nano::wallet_id, std::shared_ptr<nano::wallet>> get_wallets ();
 	nano::network_params & network_params;
 	std::function<void (bool)> observer;