test: speed up the slowest functional tests without reducing coverage

test/functional/feature_block.py

@@ -22,6 +22,7 @@
     CTxIn,
     CTxOut,
     SEQUENCE_FINAL,
+    msg_ping,
     uint256_from_compact,
     uint256_from_str,
 )
@@ -81,6 +82,28 @@ def normal_serialize(self):
 DUPLICATE_COINBASE_SCRIPT_SIG = b'\x01\x78'  # Valid for block at height 120
 
 
+class P2PBlockStore(P2PDataStore):
+    def wait_for_disconnect(self, *, timeout=60):
+        # dashd's socket thread closes a peer marked for disconnection only once it wakes up,
+        # which takes up to its 500ms poll timeout when no data comes in. Keep pinging the node
+        # so it wakes up and hangs up as soon as it has decided to; the hang-up is still awaited.
+        # At most one ping per 50ms, so a node that wrongly stays connected is not flooded.
+        last_ping = 0
+
+        def disconnected():
+            nonlocal last_ping
+            if not self.is_connected:
+                return True
+            if time.monotonic() - last_ping >= 0.05:
+                last_ping = time.monotonic()
+                try:
+                    self.send_message(msg_ping(nonce=0))
+                except IOError:
+                    pass  # closed in the meantime; seen on the next check
+            return False
+        self.wait_until(disconnected, timeout=timeout, check_connected=False)
+
+
 class FullBlockTest(BitcoinTestFramework):
     def set_test_params(self):
         self.num_nodes = 1
@@ -1278,7 +1301,7 @@ def run_test(self):
         # Don't use v2transport for the large reorg, which is too slow with the unoptimized python ChaCha20 implementation
         if self.options.v2transport:
             self.nodes[0].disconnect_p2ps()
-            self.helper_peer = self.nodes[0].add_p2p_connection(P2PDataStore(), supports_v2_p2p=False)
+            self.helper_peer = self.nodes[0].add_p2p_connection(P2PBlockStore(), supports_v2_p2p=False)
         self.log.info("Test a re-org of ~2 days' worth of blocks (1088 blocks)")
 
         self.move_tip(88)
@@ -1419,7 +1442,7 @@ def bootstrap_p2p(self, timeout=10):
         """Add a P2P connection to the node.
 
         Helper to connect and wait for version handshake."""
-        self.helper_peer = self.nodes[0].add_p2p_connection(P2PDataStore())
+        self.helper_peer = self.nodes[0].add_p2p_connection(P2PBlockStore())
         # We need to wait for the initial getheaders from the peer before we
         # start populating our blockstore. If we don't, then we may run ahead
         # to the next subtest before we receive the getheaders. We'd then send

test/functional/feature_governance.py

@@ -58,6 +58,26 @@ def check_superblock(self):
 
         assert_equal(payments_found, 2)
 
+    def sync_governance_signers(self, nodes):
+        # Masternodes create and vote on triggers only from the UpdatedBlockTip validation callback,
+        # so once every node is at the same tip and has drained its validation queue they are done
+        # acting on that tip.
+        self.sync_blocks(nodes)
+        for node in nodes:
+            node.syncwithvalidationinterfacequeue()
+
+    @staticmethod
+    def valid_triggers(node):
+        return set(node.gobject("list", "valid", "triggers").keys())
+
+    def count_own_trigger_votes(self, mns, trigger_hash):
+        # Each masternode's own vote is visible on its own node, without relying on relay.
+        count = 0
+        for mn in mns:
+            collateral = self.nodes[0].protx("info", mn.proTxHash)
+            count += len(mn.get_node(self).gobject("getcurrentvotes", trigger_hash, collateral["collateralHash"], str(collateral["collateralIndex"])))
+        return count
+
     def run_test(self):
         self.log.info("Start testing...")
         governance_info = self.nodes[0].getgovernanceinfo()
@@ -212,8 +232,9 @@ def run_test(self):
         self.wait_until(lambda: len(isolated.gobject("list", "valid", "triggers")) == 1, timeout=5)
         isolated_trigger_hash = list(isolated.gobject("list", "valid", "triggers").keys())[0]
         self.wait_until(lambda: list(isolated.gobject("list", "valid", "triggers").values())[0]['YesCount'] == 1, timeout=5)
-        more_votes = self.wait_until(lambda: list(isolated.gobject("list", "valid", "triggers").values())[0]['YesCount'] > 1, timeout=5, do_assert=False)
-        assert_equal(more_votes, False)
+        # The isolated node has no peers, so its own signer is the only possible source of more votes
+        self.sync_governance_signers([isolated])
+        assert_equal(list(isolated.gobject("list", "valid", "triggers").values())[0]['YesCount'], 1)
         # Isolated node created a trigger and voted YES for it (16 votes total)
         assert_equal(isolated.gobject("count")["votes"], 16)
         # Non-isolated nodes don't see this (still 15 votes total)
@@ -227,8 +248,11 @@ def run_test(self):
         self.check_superblockbudget(False)
 
         self.log.info("The 'winner' should submit new trigger and vote for it, but it's isolated so no triggers should be found")
-        has_trigger = self.wait_until(lambda: len(self.nodes[0].gobject("list", "valid", "triggers")) >= 1, timeout=5, do_assert=False)
-        assert_equal(has_trigger, False)
+        non_isolated = [node for node in self.nodes if node is not isolated]
+        non_isolated_mns = [mn for mn in self.mninfo if mn.nodeIdx != payee_idx]
+        self.sync_governance_signers(non_isolated)
+        for node in non_isolated:
+            assert_equal(self.valid_triggers(node), set())
 
         self.log.info("Move 1 block inside the Superblock maturity window on non-isolated nodes")
         self.bump_mocktime(1)
@@ -238,8 +262,11 @@ def run_test(self):
         self.wait_until(lambda: len(self.nodes[0].gobject("list", "valid", "triggers")) == 1, timeout=5)
         winning_trigger_hash = list(self.nodes[0].gobject("list", "valid", "triggers").keys())[0]
         self.wait_until(lambda: list(self.nodes[0].gobject("list", "valid", "triggers").values())[0]['YesCount'] == 1, timeout=5)
-        more_votes = self.wait_until(lambda: list(self.nodes[0].gobject("list", "valid", "triggers").values())[0]['YesCount'] > 1, timeout=5, do_assert=False)
-        assert_equal(more_votes, False)
+        self.sync_governance_signers(non_isolated)
+        # Every masternode must know the trigger before the next block so that it votes for it there
+        self.wait_until(lambda: all(winning_trigger_hash in self.valid_triggers(node) for node in non_isolated), timeout=5)
+        # Only the trigger's creator may have voted on it at this height
+        assert_equal(self.count_own_trigger_votes(non_isolated_mns, winning_trigger_hash), 1)
         # Non-isolated node created a trigger and voted YES for it (16 votes total)
         assert_equal(self.nodes[0].gobject("count")["votes"], 16)
         # Isolated node don't see this (still 16 votes total)
@@ -258,8 +285,17 @@ def run_test(self):
 
         self.log.info("Every non-isolated MN should vote for the same trigger now, no new triggers should be created")
         self.wait_until(lambda: list(self.nodes[0].gobject("list", "valid", "triggers").values())[0]['YesCount'] == self.mn_count - 1, timeout=5)
-        more_triggers = self.wait_until(lambda: len(self.nodes[0].gobject("list", "valid", "triggers")) > 1, timeout=5, do_assert=False)
-        assert_equal(more_triggers, False)
+        self.sync_governance_signers(non_isolated)
+        for node in non_isolated:
+            assert self.valid_triggers(node) <= {winning_trigger_hash}
+
+        # Let these votes finish relaying before the isolated node reconnects: if it is offered a vote
+        # before the trigger itself, the vote is only kept as an orphan and may be dropped.
+        self.wait_until(lambda: all(node.gobject("get", winning_trigger_hash)["FundingResult"]["YesCount"] == self.mn_count - 1 for node in non_isolated), timeout=5)
+        for node in non_isolated:
+            # Run the pending once-a-second relay flush now; mockscheduler returns after a validation queue
+            # sync, which is scheduled after the advanced relay task and so runs after it
+            node.mockscheduler(1)
         # All 4 non-isolated nodes voted YES for a trigger created by a non-isolated node earlier (19 votes total)
         assert_equal(self.nodes[0].gobject("count")["votes"], 19)
         # Isolated node don't see this (still 16 votes total)
@@ -288,8 +324,22 @@ def sync_gov(node):
         self.log.info("Should see two triggers now")
         self.wait_until(lambda: len(isolated.gobject("list", "valid", "triggers")) == 2, timeout=5)
         self.wait_until(lambda: len(self.nodes[0].gobject("list", "valid", "triggers")) == 2, timeout=5)
-        more_triggers = self.wait_until(lambda: len(self.nodes[0].gobject("list", "valid", "triggers")) > 2, timeout=5, do_assert=False)
-        assert_equal(more_triggers, False)
+        # Every masternode must know both triggers before the next block so that it votes on them there
+        both_triggers = {winning_trigger_hash, isolated_trigger_hash}
+        self.wait_until(lambda: all(both_triggers <= self.valid_triggers(node) for node in self.nodes), timeout=5)
+
+        def yes_votes_known(node, trigger_hash, yes_count):
+            # A node that missed votes during the re-sync fetches them from its sync tick, which acts only once
+            # MASTERNODE_SYNC_TICK_SECONDS of mocktime have passed since it last did, so keep mocktime moving
+            self.bump_mocktime(1)
+            return node.gobject("get", trigger_hash)["FundingResult"]["YesCount"] == yes_count
+
+        for node in self.nodes:
+            self.wait_until(lambda: yes_votes_known(node, winning_trigger_hash, self.mn_count - 1), timeout=5)
+            self.wait_until(lambda: yes_votes_known(node, isolated_trigger_hash, 1), timeout=5)
+        self.sync_governance_signers(self.nodes)
+        for node in self.nodes:
+            assert_equal(self.valid_triggers(node), both_triggers)
 
         self.log.info("Move another block inside the Superblock maturity window")
         self.bump_mocktime(1)

test/functional/feature_llmq_data_recovery.py

@@ -3,8 +3,11 @@
 # Distributed under the MIT software license, see the accompanying
 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
+from contextlib import ExitStack
+import struct
 import time
 from typing import List, Optional
+from test_framework.messages import hash256, ser_string
 from test_framework.p2p import logger
 from test_framework.test_framework import (
     DashTestFramework,
@@ -24,6 +27,9 @@
 llmq_test_v17 = 102
 llmq_type_strings = {llmq_test: 'llmq_test', llmq_test_v17: 'llmq_test_v17'}
 
+# Logged by every block's recovery trigger, unless -llmq-data-recovery=0 turns it into a no-op
+RECOVERY_TRIGGER_LOG = "TriggerQuorumDataRecoveryThreads -- Process block"
+
 
 class QuorumDataRecoveryTest(DashTestFramework):
     def add_options(self, parser):
@@ -56,7 +62,11 @@ def restart_mns(self, mns=None, exclude=None, reindex=False, qvvec_sync=None, qd
 
         if qdata_recovery_enabled:
             # trigger recovery threads and wait for them to start
-            self.generate(self.nodes[0], 1, sync_fun=self.no_op)
+            restarted = [mn for mn in (self.mninfo if mns is None else mns) if mn not in (exclude or [])]
+            log_positions = self.debug_log_positions(restarted)
+            block_hash = self.generate(self.nodes[0], 1, sync_fun=self.no_op)[0]
+            # The negative check in run_test relies on this message, so make sure it is the one the trigger logs
+            self.wait_for_log_line(restarted, log_positions, f"{RECOVERY_TRIGGER_LOG} {block_hash}")
 
             self.bump_mocktime(self.quorum_data_thread_request_timeout_seconds + 1)
             time.sleep(1)
@@ -75,6 +85,24 @@ def test_mns(self, quorum_type_in, quorum_hash_in, valid_mns=None, all_mns=None,
                 assert not self.test_mn_quorum_data(mn, quorum_type_in, quorum_hash_in, test_secret, False)
         self.wait_for_quorum_data(valid_mns, quorum_type_in, quorum_hash_in, test_secret, expect_secret, recover, timeout)
 
+    def debug_log_positions(self, mns):
+        return {mn.proTxHash: mn.get_node(self).debug_log_size(encoding="utf-8") for mn in mns}
+
+    def wait_for_log_line(self, mns, log_positions, *parts):
+        """Wait until every one of mns logged a line containing all parts since log_positions was taken"""
+        def has_line(mn):
+            with open(mn.get_node(self).debug_log_path, encoding="utf-8", errors="replace") as debug_log:
+                debug_log.seek(log_positions[mn.proTxHash])
+                return any(all(part in line for part in parts) for line in debug_log)
+
+        self.wait_until(lambda: all(has_line(mn) for mn in mns))
+
+    def sync_validation(self):
+        """Every node has the same tip and has run its validation callbacks for it"""
+        self.sync_blocks()
+        for node in self.nodes:
+            node.syncwithvalidationinterfacequeue()
+
     def get_mn(self, protx_hash) -> Optional[MasternodeInfo]:
         for mn in self.mninfo: # type: MasternodeInfo
             if mn.proTxHash == protx_hash:
@@ -95,6 +123,27 @@ def get_subset_only_in_left(self, quorum_members_left, quorum_members_right) ->
             quorum_members_subset.remove(mn)
         return quorum_members_subset
 
+    def test_no_chainlock(self, node):
+        """Mine a block and show that the quorum that has to sign it can not: every masternode tried to sign it,
+        fewer members of the signing quorum than the signing threshold hold their secret key share, each of
+        those has signed the block, and no ChainLock formed"""
+        height = node.getblockcount() + 1
+        request_id = hash256(ser_string(b"clsig") + struct.pack("<I", height))[::-1].hex()
+        log_positions = self.debug_log_positions(self.mninfo)
+        block_hash = self.generate(node, 1, sync_fun=self.sync_validation)[0]
+        # Logged before the InstantSend safety gate: a block with a transaction that is neither islocked nor old
+        # enough would not be signed, and the wait for the sig shares below would time out
+        self.wait_for_log_line(self.mninfo, log_positions, f"trying to sign {block_hash}")
+
+        signing_quorum = node.quorum("selectquorum", llmq_test, request_id)["quorumHash"]
+        signers = [mn for mn in self.mninfo if self.test_mn_quorum_data(mn, llmq_test, signing_quorum)]
+        self.log.info(f"{len(signers)} members of signing quorum {signing_quorum} hold their secret key share")
+        assert len(signers) < self.llmq_threshold
+        self.wait_for_log_line(signers, log_positions, "created sigShare", f"id={request_id}, msgHash={block_hash}")
+        self.sync_validation()
+        for n in self.nodes:
+            assert not n.getblock(block_hash)["chainlock"]
+
     def test_llmq_qvvec_sync(self, llmq_sync_entries):
         self.log.info("Test with %d -llmq-qvvec-sync option(s)" % len(llmq_sync_entries))
         for llmq_sync in llmq_sync_entries:
@@ -178,23 +227,34 @@ def run_test(self):
         # Validate all but one are invalid members now
         self.test_mns(llmq_test, quorum_hash_recover, valid_mns=[last_resort_test], all_mns=member_mns_recover_test)
         self.test_mns(llmq_test_v17, quorum_hash_recover, valid_mns=[last_resort_v17], all_mns=member_mns_recover_v17)
-        # If recovery would be enabled it would trigger after the mocktime bump / mined block
+        # If recovery would be enabled it would trigger after the mocktime bump / mined block. Recovery threads are
+        # only ever started from that trigger, which runs in the block's validation callback and logs before it
+        # starts any (restart_mns checks that it does when recovery is enabled).
         self.bump_mocktime(self.quorum_data_request_expiration_timeout + 1)
-        self.generate(node, 1, sync_fun=self.no_op)
-        time.sleep(10)
+        recovery_disabled_mns = [mn for mn in self.mninfo if mn not in exclude_members]
+        # NetQuorum::UpdatedBlockTip logs these just before it calls the trigger, so they show that the callback got
+        # that far for both blocks mined below
+        tip_height = node.getblockcount()
+        reached_trigger = [f"llmqType[{llmq_test}] h[{tip_height + i}] keeping mn quorum connections" for i in (1, 2)]
+        with ExitStack() as stack:
+            for mn in recovery_disabled_mns:
+                stack.enter_context(mn.get_node(self).assert_debug_log(reached_trigger, unexpected_msgs=[RECOVERY_TRIGGER_LOG]))
+            self.generate(node, 1, sync_fun=self.sync_validation)
+            # Mining a block should not result in a chainlock now because the responsible quorum shouldn't have
+            # enough valid members.
+            self.test_no_chainlock(node)
         # Make sure they are still invalid
         self.test_mns(llmq_test, quorum_hash_recover, valid_mns=[last_resort_test], all_mns=member_mns_recover_test)
         self.test_mns(llmq_test_v17, quorum_hash_recover, valid_mns=[last_resort_v17], all_mns=member_mns_recover_v17)
-        # Mining a block should not result in a chainlock now because the responsible quorum shouldn't have enough
-        # valid members.
-        self.wait_for_chainlocked_block(node, self.generate(node, 1, sync_fun=self.no_op)[0], False, 5)
         # Now restart with recovery enabled
         self.restart_mns(mns=recover_members, exclude=exclude_members, reindex=True, qdata_recovery_enabled=True)
         # Validate that all invalid members recover. Note: recover=True leads to mocktime bumps and mining while waiting
         # which trigger CQuorumManger::TriggerQuorumDataRecoveryThreads()
         self.test_mns(llmq_test, quorum_hash_recover, valid_mns=member_mns_recover_test, recover=True)
         self.test_mns(llmq_test_v17, quorum_hash_recover, valid_mns=member_mns_recover_v17, recover=True)
-        # Mining a block should result in a chainlock now because the quorum should be healed
+        # Mining a block should result in a chainlock now because the quorum should be healed. The tip that the
+        # recovery loop mined last may have been signed before every member held its key share again.
+        self.generate(node, 1, sync_fun=self.sync_validation)
         self.wait_for_chainlocked_block_all_nodes(node.getbestblockhash())
         logger.info("Test -llmq-qvvec-sync command line parameter")
         # Run with one type separated and then both possible (for regtest) together, both calls generate new quorums