From 84fa86574a0dc6fdaef8c789f58b7ab150c3db9d Mon Sep 17 00:00:00 2001
From: furszy <matiasfurszyfer@protonmail.com>
Date: Mon, 20 Feb 2023 20:19:02 -0300
Subject: [PATCH] index: enable block filter index parallel sync

It also adds coverage for initial sync from a particular block.
Mimicking a node restart.
---
 src/index/blockfilterindex.cpp       | 17 +++++++++++
 src/index/blockfilterindex.h         |  5 ++++
 src/test/blockfilter_index_tests.cpp | 45 ++++++++++++++++++++++++++++
 3 files changed, 67 insertions(+)

diff --git a/src/index/blockfilterindex.cpp b/src/index/blockfilterindex.cpp
index 2ccae3a221b..5b6ed8d2f4d 100644
--- a/src/index/blockfilterindex.cpp
+++ b/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(m_filter_type, *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)
 {
diff --git a/src/index/blockfilterindex.h b/src/index/blockfilterindex.h
index 983cf32206a..3ad27102b24 100644
--- a/src/index/blockfilterindex.h
+++ b/src/index/blockfilterindex.h
@@ -64,6 +64,9 @@ protected:
 
     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,
@@ -71,6 +74,8 @@ public:
 
     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;
 
diff --git a/src/test/blockfilter_index_tests.cpp b/src/test/blockfilter_index_tests.cpp
index fb3cb561b7c..0499322eb40 100644
--- a/src/test/blockfilter_index_tests.cpp
+++ b/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>
@@ -269,6 +270,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("blockfilter");
+        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;