coins: introduce thread pool in CoinsViewOverlay

diff --git a/src/coins.h b/src/coins.h
index ae7f34f4..69eee034 100644
--- a/src/coins.h
+++ b/src/coins.h
@@ -22,8 +22,11 @@
 #include <cstdint>
 
 #include <functional>
+#include <memory>
 #include <unordered_map>
 
+class ThreadPool;
+
 /**
  * A UTXO entry.
  *
@@ -569,8 +572,16 @@ private:
         return base->PeekCoin(outpoint);
     }
 
+    //! Non-null.
+    std::shared_ptr<ThreadPool> m_thread_pool;
+
 public:
-    using CCoinsViewCache::CCoinsViewCache;
+    explicit CoinsViewOverlay(CCoinsView* in_base, std::shared_ptr<ThreadPool> thread_pool,
+                              bool deterministic = false) noexcept
+        : CCoinsViewCache{in_base, deterministic}, m_thread_pool{std::move(thread_pool)}
+    {
+        Assert(m_thread_pool);
+    }
 };
 
 //! Utility function to add all of a transaction's outputs to a cache.
diff --git a/src/kernel/CMakeLists.txt b/src/kernel/CMakeLists.txt
index 541f10b3..d2a467f6 100644
--- a/src/kernel/CMakeLists.txt
+++ b/src/kernel/CMakeLists.txt
@@ -61,6 +61,7 @@ add_library(bitcoinkernel
   ../uint256.cpp
   ../util/chaintype.cpp
   ../util/check.cpp
+  ../util/exception.cpp
   ../util/expected.cpp
   ../util/feefrac.cpp
   ../util/fs.cpp
@@ -70,6 +71,7 @@ add_library(bitcoinkernel
   ../util/rbf.cpp
   ../util/signalinterrupt.cpp
   ../util/syserror.cpp
+  ../util/thread.cpp
   ../util/threadnames.cpp
   ../util/time.cpp
   ../util/tokenpipe.cpp
diff --git a/src/test/coinsviewoverlay_tests.cpp b/src/test/coinsviewoverlay_tests.cpp
index 6b20b312..fc7edfe9 100644
--- a/src/test/coinsviewoverlay_tests.cpp
+++ b/src/test/coinsviewoverlay_tests.cpp
@@ -3,6 +3,7 @@
 // file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
 #include <coins.h>
+#include <kernel/chainstatemanager_opts.h>
 #include <primitives/block.h>
 #include <primitives/transaction.h>
 #include <primitives/transaction_identifier.h>
@@ -10,17 +11,24 @@
 #include <uint256.h>
 #include <util/byte_units.h>
 #include <util/hasher.h>
+#include <util/threadpool.h>
 
 #include <boost/test/unit_test.hpp>
 
 #include <cstdint>
 #include <cstring>
+#include <memory>
 #include <ranges>
 
-BOOST_AUTO_TEST_SUITE(coinsviewoverlay_tests)
-
 namespace {
 
+std::shared_ptr<ThreadPool> MakeStartedThreadPool()
+{
+    auto pool{std::make_shared<ThreadPool>("fetch_test")};
+    pool->Start(DEFAULT_PREVOUTFETCH_THREADS);
+    return pool;
+}
+
 CBlock CreateBlock() noexcept
 {
     static constexpr auto NUM_TXS{100};
@@ -78,13 +86,15 @@ void CheckCache(const CBlock& block, const CCoinsViewCache& cache)
 
 } // namespace
 
+BOOST_AUTO_TEST_SUITE(coinsviewoverlay_tests)
+
 BOOST_AUTO_TEST_CASE(fetch_inputs_from_db)
 {
     const auto block{CreateBlock()};
     CCoinsViewDB db{{.path = "", .cache_bytes = 1_MiB, .memory_only = true}, {}};
     PopulateView(block, db);
     CCoinsViewCache main_cache{&db};
-    CoinsViewOverlay view{&main_cache};
+    CoinsViewOverlay view{&main_cache, MakeStartedThreadPool()};
     const auto& outpoint{block.vtx[1]->vin[0].prevout};
 
     BOOST_CHECK(view.HaveCoin(outpoint));
@@ -111,7 +121,7 @@ BOOST_AUTO_TEST_CASE(fetch_inputs_from_cache)
     CCoinsViewDB db{{.path = "", .cache_bytes = 1_MiB, .memory_only = true}, {}};
     CCoinsViewCache main_cache{&db};
     PopulateView(block, main_cache);
-    CoinsViewOverlay view{&main_cache};
+    CoinsViewOverlay view{&main_cache, MakeStartedThreadPool()};
     CheckCache(block, view);
 
     const auto& outpoint{block.vtx[1]->vin[0].prevout};
@@ -131,7 +141,7 @@ BOOST_AUTO_TEST_CASE(fetch_no_double_spend)
     CCoinsViewCache main_cache{&db};
     // Add all inputs as spent already in cache
     PopulateView(block, main_cache, /*spent=*/true);
-    CoinsViewOverlay view{&main_cache};
+    CoinsViewOverlay view{&main_cache, MakeStartedThreadPool()};
     for (const auto& tx : block.vtx) {
         for (const auto& in : tx->vin) {
             const auto& c{view.AccessCoin(in.prevout)};
@@ -149,7 +159,7 @@ BOOST_AUTO_TEST_CASE(fetch_no_inputs)
     const auto block{CreateBlock()};
     CCoinsViewDB db{{.path = "", .cache_bytes = 1_MiB, .memory_only = true}, {}};
     CCoinsViewCache main_cache{&db};
-    CoinsViewOverlay view{&main_cache};
+    CoinsViewOverlay view{&main_cache, MakeStartedThreadPool()};
     for (const auto& tx : block.vtx) {
         for (const auto& in : tx->vin) {
             const auto& c{view.AccessCoin(in.prevout)};
diff --git a/src/test/fuzz/coins_view.cpp b/src/test/fuzz/coins_view.cpp
index db81e190..793ea9e2 100644
--- a/src/test/fuzz/coins_view.cpp
+++ b/src/test/fuzz/coins_view.cpp
@@ -7,6 +7,7 @@
 #include <consensus/tx_check.h>
 #include <consensus/tx_verify.h>
 #include <consensus/validation.h>
+#include <kernel/chainstatemanager_opts.h>
 #include <kernel/cs_main.h>
 #include <policy/policy.h>
 #include <primitives/transaction.h>
@@ -17,6 +18,7 @@
 #include <test/util/setup_common.h>
 #include <txdb.h>
 #include <util/hasher.h>
+#include <util/threadpool.h>
 
 #include <cassert>
 #include <algorithm>
@@ -84,6 +86,12 @@ public:
 
     using CCoinsViewCache::CCoinsViewCache;
 };
+
+// Reuse a single global thread pool across fuzz iterations. Creating and destroying a pool every
+// iteration leaks memory, since iterations can run faster than the OS can tear down the threads.
+std::shared_ptr<ThreadPool> g_thread_pool{std::make_shared<ThreadPool>("view_fuzz")};
+Mutex g_thread_pool_mutex;
+
 } // namespace
 
 void initialize_coins_view()
@@ -376,10 +384,10 @@ FUZZ_TARGET(coins_view_db, .init = initialize_coins_view)
 // This allows us to exercise all methods on a CoinsViewOverlay, while also
 // ensuring that nothing can mutate the underlying cache until Flush or Sync is
 // called.
-FUZZ_TARGET(coins_view_overlay, .init = initialize_coins_view)
+FUZZ_TARGET(coins_view_overlay, .init = initialize_coins_view) EXCLUSIVE_LOCKS_REQUIRED(!g_thread_pool_mutex)
 {
     FuzzedDataProvider fuzzed_data_provider{buffer.data(), buffer.size()};
     MutationGuardCoinsViewCache backend_cache{&CoinsViewEmpty::Get(), /*deterministic=*/true};
-    CoinsViewOverlay coins_view_cache{&backend_cache, /*deterministic=*/true};
+    CoinsViewOverlay coins_view_cache{&backend_cache, g_thread_pool, /*deterministic=*/true};
     TestCoinsView(fuzzed_data_provider, coins_view_cache, &backend_cache);
 }
diff --git a/src/test/fuzz/coinscache_sim.cpp b/src/test/fuzz/coinscache_sim.cpp
index 15ece2e4..24a0327d 100644
--- a/src/test/fuzz/coinscache_sim.cpp
+++ b/src/test/fuzz/coinscache_sim.cpp
@@ -4,10 +4,13 @@
 
 #include <coins.h>
 #include <crypto/sha256.h>
+#include <kernel/chainstatemanager_opts.h>
 #include <primitives/transaction.h>
 #include <test/fuzz/FuzzedDataProvider.h>
 #include <test/fuzz/fuzz.h>
 #include <test/fuzz/util.h>
+#include <test/util/setup_common.h>
+#include <util/threadpool.h>
 
 #include <cassert>
 #include <cstdint>
@@ -182,10 +185,22 @@ public:
     }
 };
 
+// Reuse a single global thread pool across fuzz iterations. Creating and destroying a pool every
+// iteration leaks memory, since iterations can run faster than the OS can tear down the threads.
+std::shared_ptr<ThreadPool> g_thread_pool{std::make_shared<ThreadPool>("cache_fuzz")};
+Mutex g_thread_pool_mutex;
+
+void StartPoolIfNeeded() EXCLUSIVE_LOCKS_REQUIRED(!g_thread_pool_mutex)
+{
+    LOCK(g_thread_pool_mutex);
+    if (!g_thread_pool->WorkersCount()) g_thread_pool->Start(DEFAULT_PREVOUTFETCH_THREADS);
+}
+
 } // namespace
 
-FUZZ_TARGET(coinscache_sim)
+FUZZ_TARGET(coinscache_sim, .init = [] { static auto setup{MakeNoLogFileContext<>()}; }) EXCLUSIVE_LOCKS_REQUIRED(!g_thread_pool_mutex)
 {
+    StartPoolIfNeeded();
     /** Precomputed COutPoint and CCoins values. */
     static const PrecomputedData data;
 
@@ -372,7 +387,7 @@ FUZZ_TARGET(coinscache_sim)
                     if (provider.ConsumeBool()) {
                         caches.emplace_back(new CCoinsViewCache(&*caches.back(), /*deterministic=*/true));
                     } else {
-                        caches.emplace_back(new CoinsViewOverlay(&*caches.back(), /*deterministic=*/true));
+                        caches.emplace_back(new CoinsViewOverlay(&*caches.back(), g_thread_pool, /*deterministic=*/true));
                     }
                     // Apply to simulation data.
                     sim_caches[caches.size()].Wipe();
diff --git a/src/validation.cpp b/src/validation.cpp
index 87cf646b..fd1f4404 100644
--- a/src/validation.cpp
+++ b/src/validation.cpp
@@ -60,6 +60,7 @@
 #include <util/signalinterrupt.h>
 #include <util/strencodings.h>
 #include <util/string.h>
+#include <util/threadpool.h>
 #include <util/time.h>
 #include <util/trace.h>
 #include <util/translation.h>
@@ -1859,11 +1860,16 @@ CoinsViews::CoinsViews(DBParams db_params, CoinsViewOptions options)
     : m_dbview{std::move(db_params), std::move(options)},
       m_catcherview(&m_dbview) {}
 
-void CoinsViews::InitCache()
+void CoinsViews::InitCache(int32_t prevoutfetch_threads)
 {
     AssertLockHeld(::cs_main);
     m_cacheview = std::make_unique<CCoinsViewCache>(&m_catcherview);
-    m_connect_block_view = std::make_unique<CoinsViewOverlay>(&*m_cacheview);
+    auto thread_pool{std::make_shared<ThreadPool>("prevout")};
+    if (prevoutfetch_threads > 0) {
+        thread_pool->Start(prevoutfetch_threads);
+        LogInfo("Block input prevout fetching uses %d additional threads", prevoutfetch_threads);
+    }
+    m_connect_block_view = std::make_unique<CoinsViewOverlay>(&*m_cacheview, std::move(thread_pool));
 }
 
 Chainstate::Chainstate(
@@ -1939,7 +1945,7 @@ void Chainstate::InitCoinsCache(size_t cache_size_bytes)
     AssertLockHeld(::cs_main);
     assert(m_coins_views != nullptr);
     m_coinstip_cache_size_bytes = cache_size_bytes;
-    m_coins_views->InitCache();
+    m_coins_views->InitCache(m_chainman.m_options.prevoutfetch_threads_num);
 }
 
 // Lock-free: depends on `m_cached_is_ibd`, which is latched by `UpdateIBDStatus()`.
diff --git a/src/validation.h b/src/validation.h
index 2472dd60..9728e6b0 100644
--- a/src/validation.h
+++ b/src/validation.h
@@ -506,7 +506,7 @@ public:
     CoinsViews(DBParams db_params, CoinsViewOptions options);
 
     //! Initialize the CCoinsViewCache member.
-    void InitCache() EXCLUSIVE_LOCKS_REQUIRED(::cs_main);
+    void InitCache(int32_t prevoutfetch_threads) EXCLUSIVE_LOCKS_REQUIRED(::cs_main);
 };
 
 enum class CoinsCacheSizeState
diff --git a/test/functional/feature_block.py b/test/functional/feature_block.py
index 5600eeec..6e3f342c 100755
--- a/test/functional/feature_block.py
+++ b/test/functional/feature_block.py
@@ -92,6 +92,9 @@ class FullBlockTest(BitcoinTestFramework):
         self.setup_clean_chain = True
         self.extra_args = [[
             '-testactivationheight=bip34@2',
+            # Override the functional-test default of 1 thread to exercise the multi-threaded
+            # prevout prefetching path in this block-heavy test.
+            '-prevoutfetchthreads=8',
         ]]
 
     def add_options(self, parser):
diff --git a/test/functional/feature_proxy.py b/test/functional/feature_proxy.py
index fa4e966e..37a6e2f4 100755
--- a/test/functional/feature_proxy.py
+++ b/test/functional/feature_proxy.py
@@ -135,6 +135,9 @@ class ProxyTest(BitcoinTestFramework):
         if self.have_unix_sockets:
             args[5] = ['-listen', f'-proxy=unix:{socket_path}']
             args[6] = ['-listen', f'-onion=unix:{socket_path}']
+        # This test launches many nodes; disable prevout prefetching so we don't spin up a
+        # thread pool for each one.
+        args = [a + ['-prevoutfetchthreads=0'] for a in args]
         self.add_nodes(self.num_nodes, extra_args=args)
         self.start_nodes()
 
diff --git a/test/functional/test_framework/util.py b/test/functional/test_framework/util.py
index 7d064d68..a7f3252c 100644
--- a/test/functional/test_framework/util.py
+++ b/test/functional/test_framework/util.py
@@ -566,6 +566,8 @@ def write_config(config_path, *, n, chain, extra_config="", disable_autoconnect=
         #  nMaxConnections = available_fds - min_required_fds = 256 - 161 = 94;
         f.write("maxconnections=94\n")
         f.write("par=" + str(min(2, os.cpu_count())) + "\n")
+        # Use a single prevoutfetch worker thread to keep per-node resource usage low.
+        f.write("prevoutfetchthreads=1\n")
         f.write(extra_config)