validation: collect block inputs in CoinsViewOverlay before ConnectBlock

diff --git a/src/coins.cpp b/src/coins.cpp
index c403e006..69ad2648 100644
--- a/src/coins.cpp
+++ b/src/coins.cpp
@@ -5,11 +5,16 @@
 #include <coins.h>
 
 #include <consensus/consensus.h>
+#include <primitives/block.h>
 #include <random.h>
 #include <uint256.h>
 #include <util/log.h>
+#include <util/threadpool.h>
 #include <util/trace.h>
 
+#include <ranges>
+#include <unordered_set>
+
 TRACEPOINT_SEMAPHORE(utxocache, add);
 TRACEPOINT_SEMAPHORE(utxocache, spent);
 TRACEPOINT_SEMAPHORE(utxocache, uncache);
@@ -361,6 +366,31 @@ void CCoinsViewCache::SanityCheck() const
     assert(recomputed_usage == cachedCoinsUsage);
 }
 
+CCoinsViewCache::ResetGuard CoinsViewOverlay::StartFetching(const CBlock& block LIFETIMEBOUND) noexcept
+{
+    Assert(m_inputs.empty());
+    Assert(m_input_head.load(std::memory_order_relaxed) == 0);
+    Assert(m_input_tail == 0);
+    if (const auto workers_count{m_thread_pool->WorkersCount()}; workers_count > 0) {
+        // Loop through the block inputs and set their prevouts in the queue.
+        // Filter inputs that spend outputs created earlier in the same block. These outputs will be created
+        // directly in the cache from the tx that creates them, so they will not be requested from a base view.
+        std::unordered_set<Txid, SaltedTxidHasher> earlier_txids;
+        earlier_txids.reserve(block.vtx.size());
+        for (const auto& tx : block.vtx | std::views::drop(1)) {
+            for (const auto& input : tx->vin) {
+                if (!earlier_txids.contains(input.prevout.hash)) m_inputs.emplace_back(input.prevout);
+            }
+            earlier_txids.emplace(tx->GetHash());
+        }
+        // Only process inputs if we have something to fetch.
+        if (m_inputs.size()) {
+            while (ProcessInput()) {}
+        }
+    }
+    return CreateResetGuard();
+}
+
 static const uint64_t MIN_TRANSACTION_OUTPUT_WEIGHT{WITNESS_SCALE_FACTOR * ::GetSerializeSize(CTxOut())};
 static const uint64_t MAX_OUTPUTS_PER_BLOCK{MAX_BLOCK_WEIGHT / MIN_TRANSACTION_OUTPUT_WEIGHT};
 
diff --git a/src/coins.h b/src/coins.h
index 69eee034..23c5fa5f 100644
--- a/src/coins.h
+++ b/src/coins.h
@@ -11,6 +11,7 @@
 #include <core_memusage.h>
 #include <memusage.h>
 #include <primitives/transaction.h>
+#include <primitives/transaction_identifier.h>
 #include <serialize.h>
 #include <support/allocators/pool.h>
 #include <uint256.h>
@@ -21,10 +22,15 @@
 #include <cassert>
 #include <cstdint>
 
+#include <atomic>
 #include <functional>
 #include <memory>
+#include <optional>
 #include <unordered_map>
+#include <utility>
+#include <vector>
 
+class CBlock;
 class ThreadPool;
 
 /**
@@ -418,7 +424,7 @@ protected:
      * Discard all modifications made to this cache without flushing to the base view.
      * This can be used to efficiently reuse a cache instance across multiple operations.
      */
-    void Reset() noexcept;
+    virtual void Reset() noexcept;
 
     /* Fetch the coin from base. Used for cache misses in FetchCoin. */
     virtual std::optional<Coin> FetchCoinFromBase(const COutPoint& outpoint) const;
@@ -567,14 +573,72 @@ private:
 class CoinsViewOverlay : public CCoinsViewCache
 {
 private:
+    //! The latest input not yet being fetched. Workers atomically increment this when fetching.
+    std::atomic_uint32_t m_input_head{0};
+    //! The latest input not yet accessed by a consumer. Only the main thread increments this.
+    mutable uint32_t m_input_tail{0};
+
+    //! The inputs of the block which is being fetched.
+    struct InputToFetch {
+        //! The outpoint of the input to fetch.
+        const COutPoint& outpoint;
+        //! The coin that workers will fetch and main thread will insert into cache.
+        //! Mutable so it can be moved in FetchCoinFromBase.
+        mutable std::optional<Coin> coin{std::nullopt};
+
+        explicit InputToFetch(const COutPoint& o LIFETIMEBOUND) noexcept : outpoint{o} {}
+    };
+    std::vector<InputToFetch> m_inputs{};
+
+    /**
+     * Claim and fetch the next input in the queue.
+     *
+     * @return true if an input prevout was fetched
+     * @return false if there are no more input prevouts in the queue to fetch
+     */
+    bool ProcessInput() noexcept
+    {
+        const auto i{m_input_head.fetch_add(1, std::memory_order_relaxed)};
+        if (i >= m_inputs.size()) return false;
+
+        auto& input{m_inputs[i]};
+        input.coin = base->PeekCoin(input.outpoint);
+        return true;
+    }
+
+    //! Clear fetching data.
+    void StopFetching() noexcept
+    {
+        m_inputs.clear();
+        m_input_head.store(0, std::memory_order_relaxed);
+        m_input_tail = 0;
+    }
+
     std::optional<Coin> FetchCoinFromBase(const COutPoint& outpoint) const override
     {
+        // This assumes ConnectBlock accesses all inputs in the same order as
+        // they are added to m_inputs in StartFetching.
+        if (m_input_tail < m_inputs.size() && m_inputs[m_input_tail].outpoint == outpoint) {
+            // We advance the tail since the input is cached and not accessed through this method again.
+            auto& input{m_inputs[m_input_tail++]};
+            // We can move the coin since we won't access this input again.
+            return std::move(input.coin);
+        }
+
+        // We will only get here for BIP30 checks, an invalid block, or if the threadpool has not been started.
         return base->PeekCoin(outpoint);
     }
 
     //! Non-null.
     std::shared_ptr<ThreadPool> m_thread_pool;
 
+protected:
+    void Reset() noexcept override
+    {
+        StopFetching();
+        CCoinsViewCache::Reset();
+    }
+
 public:
     explicit CoinsViewOverlay(CCoinsView* in_base, std::shared_ptr<ThreadPool> thread_pool,
                               bool deterministic = false) noexcept
@@ -582,6 +646,9 @@ public:
     {
         Assert(m_thread_pool);
     }
+
+    //! Start fetching inputs from block.
+    [[nodiscard]] ResetGuard StartFetching(const CBlock& block LIFETIMEBOUND) noexcept;
 };
 
 //! Utility function to add all of a transaction's outputs to a cache.
diff --git a/src/test/fuzz/coins_view.cpp b/src/test/fuzz/coins_view.cpp
index 793ea9e2..62e577fd 100644
--- a/src/test/fuzz/coins_view.cpp
+++ b/src/test/fuzz/coins_view.cpp
@@ -386,6 +386,7 @@ FUZZ_TARGET(coins_view_db, .init = initialize_coins_view)
 // called.
 FUZZ_TARGET(coins_view_overlay, .init = initialize_coins_view) EXCLUSIVE_LOCKS_REQUIRED(!g_thread_pool_mutex)
 {
+    SeedRandomStateForTest(SeedRand::ZEROS); // for SaltedTxidHasher
     FuzzedDataProvider fuzzed_data_provider{buffer.data(), buffer.size()};
     MutationGuardCoinsViewCache backend_cache{&CoinsViewEmpty::Get(), /*deterministic=*/true};
     CoinsViewOverlay coins_view_cache{&backend_cache, g_thread_pool, /*deterministic=*/true};
diff --git a/src/validation.cpp b/src/validation.cpp
index fd1f4404..d0083444 100644
--- a/src/validation.cpp
+++ b/src/validation.cpp
@@ -3052,8 +3052,8 @@ bool Chainstate::ConnectTip(
     LogDebug(BCLog::BENCH, "  - Load block from disk: %.2fms\n",
              Ticks<MillisecondsDouble>(time_2 - time_1));
     {
-        CCoinsViewCache& view{*m_coins_views->m_connect_block_view};
-        const auto reset_guard{view.CreateResetGuard()};
+        CoinsViewOverlay& view{*m_coins_views->m_connect_block_view};
+        const auto reset_guard{view.StartFetching(*block_to_connect)};
         bool rv = ConnectBlock(*block_to_connect, state, pindexNew, view);
         if (m_chainman.m_options.signals) {
             m_chainman.m_options.signals->BlockChecked(block_to_connect, state);