Merge #12206: qa: Sync with validationinterface queue in sync_mempools

fa1e69e qa: Sync with validationinterface queue in sync_mempools (MarcoFalke)

Pull request description:

  Commit e545dedf72 moved `TransactionAddedToMempool` to the background scheduler thread. Thus, adding a transaction to the mempool will no longer add it to the wallet immediately. Functional tests, that `sync_mempools` and then call into wallet rpcs will race against the scheduler thread.

  Fix that race by flushing the scheduler queue.

  Fixes #12205; Fixes #12171;
  References #9584;

Tree-SHA512: 14d99cff9c4756de9fad412f04e6d8e25bb9a0938f24ed8348de79df5b4ee67763dac5214b1a69e77e60787d81ee642976d1482b1b5637edfc4892a238ed22af

src/rpc/blockchain.cpp

@@ -24,6 +24,7 @@
 #include <util.h>
 #include <utilstrencodings.h>
 #include <hash.h>
+#include <validationinterface.h>
 #include <warnings.h>
 
 #include <stdint.h>
@@ -323,6 +324,21 @@ UniValue waitforblockheight(const JSONRPCRequest& request)
     return ret;
 }
 
+UniValue syncwithvalidationinterfacequeue(const JSONRPCRequest& request)
+{
+    if (request.fHelp || request.params.size() > 0) {
+        throw std::runtime_error(
+            "syncwithvalidationinterfacequeue\n"
+            "\nWaits for the validation interface queue to catch up on everything that was there when we entered this function.\n"
+            "\nExamples:\n"
+            + HelpExampleCli("syncwithvalidationinterfacequeue","")
+            + HelpExampleRpc("syncwithvalidationinterfacequeue","")
+        );
+    }
+    SyncWithValidationInterfaceQueue();
+    return NullUniValue;
+}
+
 UniValue getdifficulty(const JSONRPCRequest& request)
 {
     if (request.fHelp || request.params.size() != 0)
@@ -1628,6 +1644,7 @@ static const CRPCCommand commands[] =
     { "hidden",             "waitfornewblock",        &waitfornewblock,        {"timeout"} },
     { "hidden",             "waitforblock",           &waitforblock,           {"blockhash","timeout"} },
     { "hidden",             "waitforblockheight",     &waitforblockheight,     {"height","timeout"} },
+    { "hidden",             "syncwithvalidationinterfacequeue", &syncwithvalidationinterfacequeue, {} },
 };
 
 void RegisterBlockchainRPCCommands(CRPCTable &t)

test/functional/test_framework/util.py

@@ -390,7 +390,7 @@ def sync_chain(rpc_connections, *, wait=1, timeout=60):
         timeout -= wait
     raise AssertionError("Chain sync failed: Best block hashes don't match")
 
-def sync_mempools(rpc_connections, *, wait=1, timeout=60):
+def sync_mempools(rpc_connections, *, wait=1, timeout=60, flush_scheduler=True):
     """
     Wait until everybody has the same transactions in their memory
     pools
@@ -402,6 +402,9 @@ def sync_mempools(rpc_connections, *, wait=1, timeout=60):
             if set(rpc_connections[i].getrawmempool()) == pool:
                 num_match = num_match + 1
         if num_match == len(rpc_connections):
+            if flush_scheduler:
+                for r in rpc_connections:
+                    r.syncwithvalidationinterfacequeue()
             return
         time.sleep(wait)
         timeout -= wait