Tie sync_blocks in RPC tests to notifier thread

qa/rpc-tests/test_framework/util.py

@@ -46,7 +46,8 @@ def str_to_b64str(string):
 
 def sync_blocks(rpc_connections, wait=1):
     """
-    Wait until everybody has the same block count
+    Wait until everybody has the same block count, and has notified
+    all internal listeners of them
     """
     while True:
         counts = [ x.getblockcount() for x in rpc_connections ]
@@ -54,6 +55,14 @@ def sync_blocks(rpc_connections, wait=1):
             break
         time.sleep(wait)
 
+    # Now that the block counts are in sync, wait for the internal
+    # notifications to finish
+    while True:
+        notified = [ x.getblockchaininfo()['fullyNotified'] for x in rpc_connections ]
+        if notified == [ True ] * len(notified):
+            break
+        time.sleep(wait)
+
 def sync_mempools(rpc_connections, wait=1):
     """
     Wait until everybody has the same transactions in their memory

src/main.cpp

@@ -3028,6 +3028,8 @@ static int64_t nTimePostConnect = 0;
 
 // Protected by cs_main
 std::map<CBlockIndex*, std::list<CTransaction>> recentlyConflictedTxs;
+uint64_t nRecentlyConflictedSequence = 0;
+uint64_t nNotifiedSequence = 0;
 
 /**
  * Connect a new block to chainActive. pblock is either NULL or a pointer to a CBlock
@@ -3083,6 +3085,7 @@ bool static ConnectTip(CValidationState& state, const CChainParams& chainparams,
     // Cache the conflicted transactions for subsequent notification.
     // Updates to connected wallets are triggered by ThreadNotifyWallets
     recentlyConflictedTxs.insert(std::make_pair(pindexNew, txConflicted));
+    nRecentlyConflictedSequence += 1;
 
     EnforceNodeDeprecation(pindexNew->nHeight);
 
@@ -3092,15 +3095,29 @@ bool static ConnectTip(CValidationState& state, const CChainParams& chainparams,
     return true;
 }
 
-std::map<CBlockIndex*, std::list<CTransaction>> DrainRecentlyConflicted()
+std::pair<std::map<CBlockIndex*, std::list<CTransaction>>, uint64_t> DrainRecentlyConflicted()
 {
+    uint64_t recentlyConflictedSequence;
     std::map<CBlockIndex*, std::list<CTransaction>> txs;
     {
         LOCK(cs_main);
+        recentlyConflictedSequence = nRecentlyConflictedSequence;
         txs.swap(recentlyConflictedTxs);
     }
 
-    return txs;
+    return std::make_pair(txs, recentlyConflictedSequence);
+}
+
+void SetChainNotifiedSequence(uint64_t recentlyConflictedSequence) {
+    assert(Params().NetworkIDString() == "regtest");
+    LOCK(cs_main);
+    nNotifiedSequence = recentlyConflictedSequence;
+}
+
+bool ChainIsFullyNotified() {
+    assert(Params().NetworkIDString() == "regtest");
+    LOCK(cs_main);
+    return nRecentlyConflictedSequence == nNotifiedSequence;
 }
 
 /**

src/main.h

@@ -512,6 +512,8 @@ uint64_t CalculateCurrentUsage();
  */
 CMutableTransaction CreateNewContextualCMutableTransaction(const Consensus::Params& consensusParams, int nHeight);
 
-std::map<CBlockIndex*, std::list<CTransaction>> DrainRecentlyConflicted();
+std::pair<std::map<CBlockIndex*, std::list<CTransaction>>, uint64_t> DrainRecentlyConflicted();
+void SetChainNotifiedSequence(uint64_t recentlyConflictedSequence);
+bool ChainIsFullyNotified();
 
 #endif // BITCOIN_MAIN_H

src/rpc/blockchain.cpp

@@ -1070,6 +1070,11 @@ UniValue getblockchaininfo(const UniValue& params, bool fHelp)
 
         obj.push_back(Pair("pruneheight",        block->nHeight));
     }
+
+    if (Params().NetworkIDString() == "regtest") {
+        obj.push_back(Pair("fullyNotified", ChainIsFullyNotified()));
+    }
+
     return obj;
 }
 

src/validationinterface.cpp

@@ -112,7 +112,7 @@ void ThreadNotifyWallets()
         // Pushed in reverse, popped in order.
         std::vector<CachedBlockData> blockStack;
         // Transactions that have been recently conflicted out of the mempool.
-        std::map<CBlockIndex*, std::list<CTransaction>> recentlyConflicted;
+        std::pair<std::map<CBlockIndex*, std::list<CTransaction>>, uint64_t> recentlyConflicted;
         // Transactions that have been recently added to the mempool.
         std::pair<std::vector<CTransaction>, uint64_t> recentlyAdded;
 
@@ -152,7 +152,7 @@ void ThreadNotifyWallets()
                 blockStack.emplace_back(
                     pindex,
                     std::make_pair(oldSproutTree, oldSaplingTree),
-                    recentlyConflicted.at(pindex));
+                    recentlyConflicted.first.at(pindex));
 
                 pindex = pindex->pprev;
             }
@@ -236,6 +236,7 @@ void ThreadNotifyWallets()
         // Update the notified sequence numbers. We only need this in regtest mode,
         // and should not lock on cs or cs_main here otherwise.
         if (chainParams.NetworkIDString() == "regtest") {
+            SetChainNotifiedSequence(recentlyConflicted.second);
             mempool.SetNotifiedSequence(recentlyAdded.second);
         }
     }