index: enable block filter index parallel sync

It also adds coverage for initial sync from a particular block.
Mimicking a node restart.

Author: furszy

src/index/blockfilterindex.cpp

@@ -299,6 +299,23 @@ bool BlockFilterIndex::Write(const BlockFilter& filter, uint32_t block_height, c
     return true;
 }
 
+std::any BlockFilterIndex::CustomProcessBlock(const interfaces::BlockInfo& block_info)
+{
+    return std::make_pair(BlockFilter(BlockFilterType::BASIC, *block_info.data, *block_info.undo_data), block_info.height);
+}
+
+bool BlockFilterIndex::CustomPostProcessBlocks(const std::any& obj)
+{
+    const auto& [filter, height] = std::any_cast<std::pair<BlockFilter, int>>(obj);
+    const uint256& header = filter.ComputeHeader(m_last_header);
+    if (!Write(filter, height, header)) {
+        LogError("Error writing filters, shutting down block filters index\n");
+        return false;
+    }
+    m_last_header = header;
+    return true;
+}
+
 [[nodiscard]] static bool CopyHeightIndexToHashIndex(CDBIterator& db_it, CDBBatch& batch,
                                                      const std::string& index_name, int height)
 {

src/index/blockfilterindex.h

@@ -64,13 +64,18 @@ class BlockFilterIndex final : public BaseIndex
 
     BaseIndex::DB& GetDB() const LIFETIMEBOUND override { return *m_db; }
 
+    std::any CustomProcessBlock(const interfaces::BlockInfo& block) override;
+    bool CustomPostProcessBlocks(const std::any& obj) override;
+
 public:
     /** Constructs the index, which becomes available to be queried. */
     explicit BlockFilterIndex(std::unique_ptr<interfaces::Chain> chain, BlockFilterType filter_type,
                               size_t n_cache_size, bool f_memory = false, bool f_wipe = false);
 
     BlockFilterType GetFilterType() const { return m_filter_type; }
 
+    bool AllowParallelSync() override { return true; }
+
     /** Get a single filter by block. */
     bool LookupFilter(const CBlockIndex* block_index, BlockFilter& filter_out) const;
 

src/test/blockfilter_index_tests.cpp

@@ -13,6 +13,7 @@
 #include <pow.h>
 #include <test/util/blockfilter.h>
 #include <test/util/setup_common.h>
+#include <util/threadpool.h>
 #include <validation.h>
 
 #include <boost/test/unit_test.hpp>
@@ -268,6 +269,50 @@ BOOST_FIXTURE_TEST_CASE(blockfilter_index_initial_sync, BuildChainTestingSetup)
     filter_index.Stop();
 }
 
+BOOST_FIXTURE_TEST_CASE(blockfilter_index_parallel_initial_sync, BuildChainTestingSetup)
+{
+    int tip_height = 100; // pre-mined blocks
+    const uint16_t MINE_BLOCKS = 650;
+    for (int round = 0; round < 2; round++) { // two rounds to test sync from genesis and from a higher block
+        // Generate blocks
+        mineBlocks(MINE_BLOCKS);
+        const CBlockIndex* tip = WITH_LOCK(::cs_main, return m_node.chainman->ActiveChain().Tip());
+        BOOST_REQUIRE(tip->nHeight == MINE_BLOCKS + tip_height);
+        tip_height = tip->nHeight;
+
+        // Init index
+        BlockFilterIndex filter_index(interfaces::MakeChain(m_node), BlockFilterType::BASIC, 1 << 20, /*f_memory=*/false);
+        BOOST_REQUIRE(filter_index.Init());
+
+        ThreadPool thread_pool;
+        thread_pool.Start(2);
+        filter_index.SetThreadPool(thread_pool);
+        filter_index.SetBlocksPerWorker(200);
+
+        // Start sync
+        BOOST_CHECK(!filter_index.BlockUntilSyncedToCurrentChain());
+        filter_index.Sync();
+        const auto& summary{filter_index.GetSummary()};
+        BOOST_CHECK(summary.synced);
+        BOOST_CHECK_EQUAL(summary.best_block_height, tip_height);
+
+        // Check that filter index has all blocks that were in the chain before it started.
+        {
+            uint256 last_header;
+            LOCK(cs_main);
+            const CBlockIndex* block_index;
+            for (block_index = m_node.chainman->ActiveChain().Genesis();
+                 block_index != nullptr;
+                 block_index = m_node.chainman->ActiveChain().Next(block_index)) {
+                CheckFilterLookups(filter_index, block_index, last_header, m_node.chainman->m_blockman);
+            }
+        }
+
+        filter_index.Interrupt();
+        filter_index.Stop();
+    }
+}
+
 BOOST_FIXTURE_TEST_CASE(blockfilter_index_init_destroy, BasicTestingSetup)
 {
     BlockFilterIndex* filter_index;