Merge #7056: refactor: extract CActiveMasternodeManager from LLMQContext (1/n, preliminary refactoring)

0114f58 fix: guard `quorumBaseBlockIndexCache` (Kittywhiskers Van Gogh)
2c6cbc9 refactor: use `gsl::not_null` in `CQuorumManager` wherever possible (Kittywhiskers Van Gogh)
8f8078f refactor: initialize `CJWalletManager` after `ActiveContext` (Kittywhiskers Van Gogh)
dac4769 refactor: move `CDKGSessionManager` to `{Active,Observer}Context` (Kittywhiskers Van Gogh)
9c61900 refactor: create stub class for watch-only mode context (Kittywhiskers Van Gogh)
b9e20f4 refactor: introduce `MakePeerManager()` helper for test setup (Kittywhiskers Van Gogh)
6b811c2 refactor: make `CDKGSessionManager` a plugable obj for `CQuorumManager` (Kittywhiskers Van Gogh)
5d54cfe refactor: couple listener registration with instantiation (Kittywhiskers Van Gogh)
263dd67 refactor: drop remaining `gArgs` usage in LLMQ code (Kittywhiskers Van Gogh)
3f41401 refactor: move `GetEnabledQuorumVvecSyncEntries()` invocation to init (Kittywhiskers Van Gogh)
16459c5 refactor: drop `QuorumDataRecoveryEnabled()` (Kittywhiskers Van Gogh)
7f32d39 refactor: drop `IsWatchQuorumsEnabled()` (Kittywhiskers Van Gogh)

Pull request description:

  ## Additional Information

  * Dependency for #7062

  * As mentioned in [bitcoin#25862](bitcoin#25862), code in `libbitcoin_kernel` should not interact with `ArgsManager` and currently, Dash-specific chainstate validation relies on it to query flags like `-watchquorums` and `-llmq-data-recovery`.

    They have been refactored to be taken as constructor arguments in the style of [bitcoin#23280](bitcoin#23280) and passed from the context to the constructor of the underlying managers. Similar changes have been extended to remaining `gArgs` usage.

  * When debugging dependent PRs, it was found that there was a potential race condition as the creation of the signer was done by one entity (`ActiveContext`) but activation was done by the parent entity (e.g. `CChainLocksHandler`). Further complicating the relationship is the introduction of a network entity (e.g. `NetInstantSend`) which could cause potential lifecycle issues.

    To simpilfy this, the entity that is responsible for creation of the signer is also responsible for registration of the signer. The functions have also been renamed to follow the convention in `CSigSharesManager` to `{Unr,R}egisterAsRecoveredSigsListener()` as it more descriptive as opposed to `Start`/`Stop()`, which is associated with spawning worker entities.

  * `CDKGSessionManager` is a dormant entity if the node is not in watch-only mode or masternode mode ([source](https://github.com/dashpay/dash/blob/bac6bfadffe6fcf323e5e0208c8619e1e68401fe/src/llmq/dkgsessionmgr.cpp#L46-L49)), to allow moving them to contexts that are mode-specific (in order to keep `LLMQContext` limited to only chainstate validation logic), it is now a connectable entity in the style of [dash#5980](#5980) (a5be37c).

    * Conversely, we now have ready access to `ChainstateManager` when constructing `CMNHFManager` so we can partially revert the change introduced in [dash#6078](#6078) (208b1c0).

  * As subsequent PRs will change the ctor arguments for `PeerManager` multiple times, a helper has been added in the form of `MakePeerManager()` to contain the changes to the helper.

  * Currently, `CJWalletManager` is initialised if `CActiveMasternodeManager` is _not_ initialised. This is fine _until_ we move `CActiveMasternodeManager` into `ActiveContext` as `ActiveContext` is initialised after `CJWalletManager` and cannot be moved upward due to hard dependencies.

    As there's no reason that `CJWalletManager` needs to be initialised early, we can move it down instead.

  ## Breaking Changes

  None expected.

  ## Checklist

  - [x] I have performed a self-review of my own code
  - [x] I have commented my code, particularly in hard-to-understand areas
  - [x] I have added or updated relevant unit/integration/functional/e2e tests
  - [x] I have made corresponding changes to the documentation **(note: N/A)**
  - [x] I have assigned this pull request to a milestone _(for repository code-owners and collaborators only)_

Top commit has no ACKs.

Tree-SHA512: d4597f24f918f2b7bc3468111acce004d3df83b2229afd658bb2b4df319369f337cc3a7ae27ed8159ea36265f8f7ad12cba91299d2b156e75b13c62ebdeedc19

Author: PastaPastaPasta

src/Makefile.am

@@ -285,6 +285,7 @@ BITCOIN_CORE_H = \
   llmq/snapshot.h \
   llmq/types.h \
   llmq/utils.h \
+  llmq/observer/context.h \
   logging.h \
   logging/timer.h \
   mapport.h \
@@ -553,6 +554,7 @@ libbitcoin_node_a_SOURCES = \
   llmq/signing_shares.cpp \
   llmq/snapshot.cpp \
   llmq/utils.cpp \
+  llmq/observer/context.cpp \
   mapport.cpp \
   masternode/active/context.cpp \
   masternode/active/notificationinterface.cpp \

src/chainlock/chainlock.cpp

@@ -63,9 +63,6 @@ CChainLocksHandler::~CChainLocksHandler()
 
 void CChainLocksHandler::Start(const llmq::CInstantSendManager& isman)
 {
-    if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
-        signer->Start();
-    }
     scheduler->scheduleEvery(
         [&]() {
             auto signer = m_signer.load(std::memory_order_acquire);
@@ -83,9 +80,6 @@ void CChainLocksHandler::Start(const llmq::CInstantSendManager& isman)
 void CChainLocksHandler::Stop()
 {
     scheduler->stop();
-    if (auto signer = m_signer.load(std::memory_order_acquire); signer) {
-        signer->Stop();
-    }
 }
 
 bool CChainLocksHandler::AlreadyHave(const CInv& inv) const

src/chainlock/signing.cpp

@@ -30,12 +30,12 @@ ChainLockSigner::ChainLockSigner(CChainState& chainstate, ChainLockSignerParent&
 
 ChainLockSigner::~ChainLockSigner() = default;
 
-void ChainLockSigner::Start()
+void ChainLockSigner::RegisterRecoveryInterface()
 {
     m_sigman.RegisterRecoveredSigsListener(this);
 }
 
-void ChainLockSigner::Stop()
+void ChainLockSigner::UnregisterRecoveryInterface()
 {
     m_sigman.UnregisterRecoveredSigsListener(this);
 }

src/chainlock/signing.h

@@ -70,8 +70,8 @@ class ChainLockSigner final : public llmq::CRecoveredSigsListener
                              llmq::CSigSharesManager& shareman, CSporkManager& sporkman, const CMasternodeSync& mn_sync);
     ~ChainLockSigner();
 
-    void Start();
-    void Stop();
+    void RegisterRecoveryInterface();
+    void UnregisterRecoveryInterface();
 
     void EraseFromBlockHashTxidMap(const uint256& hash)
         EXCLUSIVE_LOCKS_REQUIRED(!cs_signer);

src/dsnotificationinterface.cpp

@@ -78,7 +78,6 @@ void CDSNotificationInterface::UpdatedBlockTip(const CBlockIndex *pindexNew, con
 
     m_llmq_ctx->isman->UpdatedBlockTip(pindexNew);
     m_llmq_ctx->clhandler->UpdatedBlockTip(*m_llmq_ctx->isman);
-    m_llmq_ctx->qdkgsman->UpdatedBlockTip(pindexNew, fInitialDownload);
     m_llmq_ctx->qman->UpdatedBlockTip(pindexNew, m_connman, fInitialDownload);
 
     if (m_govman.IsValid()) {

src/init.cpp

@@ -94,8 +94,9 @@
 #include <instantsend/net_instantsend.h>
 #include <llmq/context.h>
 #include <llmq/dkgsessionmgr.h>
-#include <llmq/options.h>
 #include <llmq/net_signing.h>
+#include <llmq/options.h>
+#include <llmq/observer/context.h>
 #include <masternode/active/context.h>
 #include <masternode/active/notificationinterface.h>
 #include <masternode/meta.h>
@@ -258,9 +259,6 @@ void Interrupt(NodeContext& node)
     if (node.peerman) {
         node.peerman->InterruptHandlers();
     }
-    if (node.llmq_ctx) {
-        node.llmq_ctx->Interrupt();
-    }
     InterruptMapPort();
     if (node.connman)
         node.connman->Interrupt();
@@ -350,6 +348,10 @@ void PrepareShutdown(NodeContext& node)
     // CValidationInterface callbacks, flush them...
     GetMainSignals().FlushBackgroundCallbacks();
 
+    if (node.observer_ctx) {
+        UnregisterValidationInterface(node.observer_ctx.get());
+    }
+
     if (g_active_notification_interface) {
         UnregisterValidationInterface(g_active_notification_interface.get());
         g_active_notification_interface.reset();
@@ -366,6 +368,7 @@ void PrepareShutdown(NodeContext& node)
 
     // After all scheduled tasks have been flushed, destroy pointers
     // and reset all to nullptr.
+    node.observer_ctx.reset();
     node.active_ctx.reset();
     node.mn_sync.reset();
     node.sporkman.reset();
@@ -1409,9 +1412,8 @@ bool AppInitParameterInteraction(const ArgsManager& args)
     }
 
     try {
-        const bool fRecoveryEnabled{llmq::QuorumDataRecoveryEnabled()};
-        const bool fQuorumVvecRequestsEnabled{llmq::GetEnabledQuorumVvecSyncEntries().size() > 0};
-        if (!fRecoveryEnabled && fQuorumVvecRequestsEnabled) {
+        const bool fQuorumVvecRequestsEnabled{llmq::GetEnabledQuorumVvecSyncEntries(args).size() > 0};
+        if (!args.GetBoolArg("-llmq-data-recovery", llmq::DEFAULT_ENABLE_QUORUM_DATA_RECOVERY) && fQuorumVvecRequestsEnabled) {
             InitWarning(Untranslated("-llmq-qvvec-sync set but recovery is disabled due to -llmq-data-recovery=0"));
         }
     } catch (const std::invalid_argument& e) {
@@ -1956,6 +1958,9 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
     fReindex = args.GetBoolArg("-reindex", false);
     bool fReindexChainState = args.GetBoolArg("-reindex-chainstate", false);
 
+    const bool quorums_recovery = args.GetBoolArg("-llmq-data-recovery", llmq::DEFAULT_ENABLE_QUORUM_DATA_RECOVERY);
+    const bool quorums_watch = args.GetBoolArg("-watchquorums", llmq::DEFAULT_WATCH_QUORUMS);
+
     // cache size calculations
     CacheSizes cache_sizes = CalculateCacheSizes(args, g_enabled_filter_types.size());
 
@@ -2023,13 +2028,27 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
                                               args.GetBoolArg("-spentindex", DEFAULT_SPENTINDEX),
                                               args.GetBoolArg("-timestampindex", DEFAULT_TIMESTAMPINDEX),
                                               chainparams.GetConsensus(),
+                                              llmq::GetEnabledQuorumVvecSyncEntries(args),
                                               fReindexChainState,
                                               cache_sizes.block_tree_db,
                                               cache_sizes.coins_db,
                                               cache_sizes.coins,
                                               /*block_tree_db_in_memory=*/false,
                                               /*coins_db_in_memory=*/false,
                                               /*dash_dbs_in_memory=*/false,
+                                              quorums_recovery,
+                                              quorums_watch,
+                                              /*bls_threads=*/[&args]() -> int8_t {
+                                                  int8_t threads = args.GetIntArg("-parbls", llmq::DEFAULT_BLSCHECK_THREADS);
+                                                  if (threads <= 0) {
+                                                      // -parbls=0 means autodetect (number of cores - 1 validator threads)
+                                                      // -parbls=-n means "leave n cores free" (number of cores - n - 1 validator threads)
+                                                      threads += GetNumCores();
+                                                  }
+                                                  // Subtract 1 because the main thread counts towards the par threads
+                                                  return std::clamp<int8_t>(threads - 1, 0, llmq::MAX_BLSCHECK_THREADS);
+                                              }(),
+                                              args.GetIntArg("-maxrecsigsage", llmq::DEFAULT_MAX_RECOVERED_SIGS_AGE),
                                               /*shutdown_requested=*/ShutdownRequested,
                                               /*coins_error_cb=*/[]() {
                                                   uiInterface.ThreadSafeMessageBox(
@@ -2175,45 +2194,56 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
     assert(!node.dstxman);
     node.dstxman = std::make_unique<CDSTXManager>();
 
-    assert(!node.cj_walletman);
-    if (!node.mn_activeman) {
-        node.cj_walletman = CJWalletManager::make(chainman, *node.dmnman, *node.mn_metaman, *node.mempool, *node.mn_sync,
-                                                  *node.llmq_ctx->isman, !ignores_incoming_txs);
-    }
-    if (node.cj_walletman) {
-        RegisterValidationInterface(node.cj_walletman.get());
-    }
-
     assert(!node.peerman);
     node.peerman = PeerManager::make(chainparams, *node.connman, *node.addrman, node.banman.get(), *node.dstxman,
                                      chainman, *node.mempool, *node.mn_metaman, *node.mn_sync,
-                                     *node.govman, *node.sporkman, node.mn_activeman.get(), node.dmnman,
-                                     node.active_ctx, node.cj_walletman.get(), node.llmq_ctx, ignores_incoming_txs);
+                                     *node.govman, *node.sporkman, node.mn_activeman.get(), node.active_ctx, node.dmnman,
+                                     node.cj_walletman, node.llmq_ctx, node.observer_ctx, ignores_incoming_txs);
     RegisterValidationInterface(node.peerman.get());
 
     g_ds_notification_interface = std::make_unique<CDSNotificationInterface>(
         *node.connman, *node.dstxman, *node.mn_sync, *node.govman, chainman, node.dmnman, node.llmq_ctx
     );
     RegisterValidationInterface(g_ds_notification_interface.get());
 
-    // ********************************************************* Step 7c: Setup masternode mode
+    // ********************************************************* Step 7c: Setup masternode mode or watch-only mode
     assert(!node.active_ctx);
     assert(!g_active_notification_interface);
+    assert(!node.observer_ctx);
+
     node.peerman->AddExtraHandler(std::make_unique<NetInstantSend>(node.peerman.get(), *node.llmq_ctx->isman, *node.llmq_ctx->qman, chainman.ActiveChainstate()));
     node.peerman->AddExtraHandler(std::make_unique<NetSigning>(node.peerman.get(), *node.llmq_ctx->sigman));
-    if (node.mn_activeman) {
-        std::unique_ptr<CCoinJoinServer> cj_server = std::make_unique<CCoinJoinServer>(node.peerman.get(), chainman, *node.connman, *node.dmnman, *node.dstxman, *node.mn_metaman, *node.mempool, *node.mn_activeman, *node.mn_sync, *node.llmq_ctx->isman);
 
-        node.active_ctx = std::make_unique<ActiveContext>(chainman, *node.connman, *node.dmnman, *node.govman,
+    const util::DbWrapperParams dash_db_params{.path = args.GetDataDirNet(), .memory = false, .wipe = (fReindex || fReindexChainState)};
+    if (node.mn_activeman) {
+        auto cj_server = std::make_unique<CCoinJoinServer>(node.peerman.get(), chainman, *node.connman, *node.dmnman, *node.dstxman, *node.mn_metaman,
+                                                           *node.mempool, *node.mn_activeman, *node.mn_sync, *node.llmq_ctx->isman);
+        node.active_ctx = std::make_unique<ActiveContext>(*cj_server, *node.connman, *node.dmnman, *node.govman, chainman, *node.mn_metaman,
                                                           *node.mnhf_manager, *node.sporkman, *node.mempool, *node.llmq_ctx, *node.peerman,
-                                                          *node.mn_activeman, *node.mn_sync, *cj_server);
+                                                          *node.mn_activeman, *node.mn_sync, dash_db_params, quorums_watch);
         node.peerman->AddExtraHandler(std::move(cj_server));
         g_active_notification_interface = std::make_unique<ActiveNotificationInterface>(*node.active_ctx, *node.mn_activeman);
         RegisterValidationInterface(g_active_notification_interface.get());
+    } else if (quorums_watch) {
+        node.observer_ctx = std::make_unique<llmq::ObserverContext>(*node.llmq_ctx->bls_worker, *node.dmnman, *node.mn_metaman, *node.llmq_ctx->dkg_debugman,
+                                                                    *node.llmq_ctx->quorum_block_processor, *node.llmq_ctx->qman, *node.llmq_ctx->qsnapman, chainman,
+                                                                    *node.sporkman, dash_db_params);
+        RegisterValidationInterface(node.observer_ctx.get());
     }
 
     // ********************************************************* Step 7d: Setup other Dash services
 
+    assert(!node.cj_walletman);
+    if (!node.active_ctx) {
+        // Can return nullptr if built without wallet support, must check before use
+        node.cj_walletman = CJWalletManager::make(chainman, *node.dmnman, *node.mn_metaman, *node.mempool, *node.mn_sync,
+                                                  *node.llmq_ctx->isman, !ignores_incoming_txs);
+    }
+
+    if (node.cj_walletman) {
+        RegisterValidationInterface(node.cj_walletman.get());
+    }
+
     bool fLoadCacheFiles = !(fReindex || fReindexChainState) && (chainman.ActiveChain().Tip() != nullptr);
 
     if (!node.netfulfilledman->LoadCache(fLoadCacheFiles)) {
@@ -2307,7 +2337,7 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
 
     // ********************************************************* Step 10a: schedule Dash-specific tasks
 
-    node.llmq_ctx->Start(*node.peerman);
+    node.llmq_ctx->Start();
     node.peerman->StartHandlers();
     if (node.active_ctx) node.active_ctx->Start(*node.connman, *node.peerman);
 
@@ -2317,7 +2347,7 @@ bool AppInitMain(NodeContext& node, interfaces::BlockAndHeaderTipInfo* tip_info)
     node.peerman->ScheduleHandlers(*node.scheduler);
 
     if (node.mn_activeman) {
-        node.scheduler->scheduleEvery(std::bind(&llmq::CDKGSessionManager::CleanupOldContributions, std::ref(*node.llmq_ctx->qdkgsman)), std::chrono::hours{1});
+        node.scheduler->scheduleEvery(std::bind(&llmq::CDKGSessionManager::CleanupOldContributions, std::ref(*node.active_ctx->qdkgsman)), std::chrono::hours{1});
     }
 
     if (node.cj_walletman) {

src/instantsend/instantsend.h

@@ -118,7 +118,7 @@ class CInstantSendManager final : public instantsend::InstantSendSignerParent
     }
     void DisconnectSigner() { m_signer.store(nullptr, std::memory_order_release); }
 
-    instantsend::InstantSendSigner* Signer() const { return m_signer.load(); }
+    instantsend::InstantSendSigner* Signer() const { return m_signer.load(std::memory_order_acquire); }
 
 private:
     void AddNonLockedTx(const CTransactionRef& tx, const CBlockIndex* pindexMined)

src/instantsend/net_instantsend.cpp

@@ -72,18 +72,10 @@ void NetInstantSend::Start()
     }
 
     workThread = std::thread(&util::TraceThread, "isman", [this] { WorkThreadMain(); });
-
-    if (auto signer = m_is_manager.Signer(); signer) {
-        signer->Start();
-    }
 }
 
 void NetInstantSend::Stop()
 {
-    if (auto signer = m_is_manager.Signer(); signer) {
-        signer->Stop();
-    }
-
     // make sure to call Interrupt() first
     if (!workInterrupt) {
         assert(false);

src/instantsend/signing.cpp

@@ -46,12 +46,12 @@ InstantSendSigner::InstantSendSigner(CChainState& chainstate, llmq::CChainLocksH
 
 InstantSendSigner::~InstantSendSigner() = default;
 
-void InstantSendSigner::Start()
+void InstantSendSigner::RegisterRecoveryInterface()
 {
     m_sigman.RegisterRecoveredSigsListener(this);
 }
 
-void InstantSendSigner::Stop()
+void InstantSendSigner::UnregisterRecoveryInterface()
 {
     m_sigman.UnregisterRecoveredSigsListener(this);
 }

src/instantsend/signing.h

@@ -82,8 +82,8 @@ class InstantSendSigner final : public llmq::CRecoveredSigsListener
                                CTxMemPool& mempool, const CMasternodeSync& mn_sync);
     ~InstantSendSigner();
 
-    void Start();
-    void Stop();
+    void RegisterRecoveryInterface();
+    void UnregisterRecoveryInterface();
 
     void ClearInputsFromQueue(const Uint256HashSet& ids) EXCLUSIVE_LOCKS_REQUIRED(!cs_input_requests);