Merge #19208: test: move sync_blocks and sync_mempool functions to test_framework.py

cc84460c16 test: move sync_blocks and sync_mempool functions to test_framework.py (Roy Shao)

Pull request description:

  This PR moves `sync_blocks` and `sync_mempool` out from `test_framework/util.py` to `test_framework/test_framework.py` so they can take contextual information of test framework into account.

  * Change all reference callers to call functions from `test_framework.py`
  * Remove `**kwargs` which is not used
  * Take into account of `timeout_factor` when respecting timeout in function implementations.
  * Pass all tests by running `./test/functional/test_runner.py`

  fixes #18930

ACKs for top commit:
  MarcoFalke:
    ACK cc84460c16 , reviewed with --color-moved=dimmed-zebra --color-moved-ws=ignore-all-space 💫

Tree-SHA512: a79b2a3fa842fc26a7aacb834bb2aea88b3049916c0b754e60002a77ce94bb5954e0ea3b436bf268e9295efb62d721dfef263a09339a55c684ac3fda388c275e

test/functional/test_framework/test_framework.py

@@ -31,8 +31,6 @@ from .util import (
     disconnect_nodes,
     get_datadir_path,
     initialize_datadir,
-    sync_blocks,
-    sync_mempools,
 )
 
 
@@ -549,15 +547,54 @@ class BitcoinTestFramework(metaclass=BitcoinTestMetaClass):
         connect_nodes(self.nodes[1], 2)
         self.sync_all()
 
-    def sync_blocks(self, nodes=None, **kwargs):
-        sync_blocks(nodes or self.nodes, **kwargs)
+    def sync_blocks(self, nodes=None, wait=1, timeout=60):
+        """
+        Wait until everybody has the same tip.
+        sync_blocks needs to be called with an rpc_connections set that has least
+        one node already synced to the latest, stable tip, otherwise there's a
+        chance it might return before all nodes are stably synced.
+        """
+        rpc_connections = nodes or self.nodes
+        timeout = int(timeout * self.options.timeout_factor)
+        stop_time = time.time() + timeout
+        while time.time() <= stop_time:
+            best_hash = [x.getbestblockhash() for x in rpc_connections]
+            if best_hash.count(best_hash[0]) == len(rpc_connections):
+                return
+            # Check that each peer has at least one connection
+            assert (all([len(x.getpeerinfo()) for x in rpc_connections]))
+            time.sleep(wait)
+        raise AssertionError("Block sync timed out after {}s:{}".format(
+            timeout,
+            "".join("\n  {!r}".format(b) for b in best_hash),
+        ))
 
-    def sync_mempools(self, nodes=None, **kwargs):
-        sync_mempools(nodes or self.nodes, **kwargs)
+    def sync_mempools(self, nodes=None, wait=1, timeout=60, flush_scheduler=True):
+        """
+        Wait until everybody has the same transactions in their memory
+        pools
+        """
+        rpc_connections = nodes or self.nodes
+        timeout = int(timeout * self.options.timeout_factor)
+        stop_time = time.time() + timeout
+        while time.time() <= stop_time:
+            pool = [set(r.getrawmempool()) for r in rpc_connections]
+            if pool.count(pool[0]) == len(rpc_connections):
+                if flush_scheduler:
+                    for r in rpc_connections:
+                        r.syncwithvalidationinterfacequeue()
+                return
+            # Check that each peer has at least one connection
+            assert (all([len(x.getpeerinfo()) for x in rpc_connections]))
+            time.sleep(wait)
+        raise AssertionError("Mempool sync timed out after {}s:{}".format(
+            timeout,
+            "".join("\n  {!r}".format(m) for m in pool),
+        ))
 
-    def sync_all(self, nodes=None, **kwargs):
-        self.sync_blocks(nodes, **kwargs)
-        self.sync_mempools(nodes, **kwargs)
+    def sync_all(self, nodes=None):
+        self.sync_blocks(nodes)
+        self.sync_mempools(nodes)
 
     # Private helper methods. These should not be accessed by the subclass test scripts.