diff --git a/test/functional/feature_block.py b/test/functional/feature_block.py
index 9f6023d963f0..406ba0af9bb0 100755
--- a/test/functional/feature_block.py
+++ b/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
diff --git a/test/functional/feature_governance.py b/test/functional/feature_governance.py
index 6aedace59c95..e544988f9ba8 100755
--- a/test/functional/feature_governance.py
+++ b/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)
diff --git a/test/functional/feature_llmq_data_recovery.py b/test/functional/feature_llmq_data_recovery.py
index 22737c44c9bc..435783059061 100755
--- a/test/functional/feature_llmq_data_recovery.py
+++ b/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
diff --git a/test/functional/feature_llmq_signing.py b/test/functional/feature_llmq_signing.py
index ee05f657adfb..7113b244ed6b 100755
--- a/test/functional/feature_llmq_signing.py
+++ b/test/functional/feature_llmq_signing.py
@@ -49,46 +49,56 @@ def run_test(self):
         msgHash = "0000000000000000000000000000000000000000000000000000000000000002"
         msgHashConflict = "0000000000000000000000000000000000000000000000000000000000000003"
 
-        def check_sigs(hasrecsigs, isconflicting1, isconflicting2):
+        def check_sigs(hasrecsigs, isconflicting1, isconflicting2, sig_id=None):
+            sig_id = sig_id or id
             for mn in self.mninfo: # type: MasternodeInfo
-                if mn.get_node(self).quorum("hasrecsig", q_type, id, msgHash) != hasrecsigs:
+                if mn.get_node(self).quorum("hasrecsig", q_type, sig_id, msgHash) != hasrecsigs:
                     return False
-                if mn.get_node(self).quorum("isconflicting", q_type, id, msgHash) != isconflicting1:
+                if mn.get_node(self).quorum("isconflicting", q_type, sig_id, msgHash) != isconflicting1:
                     return False
-                if mn.get_node(self).quorum("isconflicting", q_type, id, msgHashConflict) != isconflicting2:
+                if mn.get_node(self).quorum("isconflicting", q_type, sig_id, msgHashConflict) != isconflicting2:
                     return False
             return True
 
         def wait_for_sigs(hasrecsigs, isconflicting1, isconflicting2, timeout):
             self.wait_until(lambda: check_sigs(hasrecsigs, isconflicting1, isconflicting2), timeout = timeout)
 
-        def assert_sigs_nochange(hasrecsigs, isconflicting1, isconflicting2, timeout):
-            assert not self.wait_until(lambda: not check_sigs(hasrecsigs, isconflicting1, isconflicting2), timeout = timeout, do_assert = False)
+        def assert_sigs_nochange(hasrecsigs, isconflicting1, isconflicting2, timeout, other_ids=()):
+            def unchanged():
+                return all(check_sigs(hasrecsigs, isconflicting1, isconflicting2, sig_id) for sig_id in (id, *other_ids))
+            assert not self.wait_until(lambda: not unchanged(), timeout = timeout, do_assert = False)
+
+        # Recovering, conflicting or removing a recovered sig can't be undone during this test without a
+        # deliberate step, so a change caused by any earlier step is still visible in a later window. A step
+        # that must cause no change is therefore checked at once, and its window is shared with the next
+        # step that must cause no change either.
 
         # Initial state
         wait_for_sigs(False, False, False, 1)
 
         # Sign first share without any optional parameter, should not result in recovered sig
         self.mninfo[0].get_node(self).quorum("sign", q_type, id, msgHash)
-        assert_sigs_nochange(False, False, False, 3)
+        assert check_sigs(False, False, False)
         # Sign second share and test optional quorumHash parameter, should not result in recovered sig
         # 1. Providing an invalid quorum hash should fail and cause no changes for sigs
         assert not self.mninfo[1].get_node(self).quorum("sign", q_type, id, msgHash, msgHash)
-        assert_sigs_nochange(False, False, False, 3)
+        assert check_sigs(False, False, False)
         # 2. Providing a valid quorum hash should succeed and cause no changes for sigss
         quorumHash = self.mninfo[1].get_node(self).quorum("selectquorum", q_type, id)["quorumHash"]
         assert self.mninfo[1].get_node(self).quorum("sign", q_type, id, msgHash, quorumHash)
-        assert_sigs_nochange(False, False, False, 3)
-        # Sign third share and test optional submit parameter if spork21 is enabled, should result in recovered sig
-        # and conflict for msgHashConflict
+        assert check_sigs(False, False, False)
         if self.options.spork21:
+            # Test optional submit parameter if spork21 is enabled
             # 1. Providing an invalid quorum hash and set submit=false, should throw an error
             assert_raises_rpc_error(-8, 'quorum not found', self.mninfo[2].get_node(self).quorum, "sign", q_type, id, msgHash, id, False)
             # 2. Providing a valid quorum hash and set submit=false, should return a valid sigShare object
             sig_share_rpc_1 = self.mninfo[2].get_node(self).quorum("sign", q_type, id, msgHash, quorumHash, False)
             sig_share_rpc_2 = self.mninfo[2].get_node(self).quorum("sign", q_type, id, msgHash, "", False)
             assert_equal(sig_share_rpc_1, sig_share_rpc_2)
-            assert_sigs_nochange(False, False, False, 3)
+        # Window shared by every step above
+        assert_sigs_nochange(False, False, False, 3)
+        # Sign third share, should result in recovered sig and conflict for msgHashConflict
+        if self.options.spork21:
             # 3. Sending the sig share received from RPC to the recovery member through P2P interface, should result
             # in a recovered sig
             sig_share = CSigShare()
@@ -134,7 +144,9 @@ def assert_sigs_nochange(hasrecsigs, isconflicting1, isconflicting2, timeout):
 
         # Mine one more quorum, so that we have 2 active ones, nothing should change
         self.mine_quorum()
-        assert_sigs_nochange(True, False, True, 3)
+        # The window for this is shared with the one after the next quorums below
+        assert check_sigs(True, False, True)
+        first_id = id
 
         # Create a recovered sig for the oldest quorum i.e. the active quorum which will be moved
         # out of the active set when a new quorum appears
@@ -162,7 +174,7 @@ def assert_sigs_nochange(hasrecsigs, isconflicting1, isconflicting2, timeout):
         # Mine 2 more quorums, so that the one used for the the recovered sig should become inactive, nothing should change
         self.mine_quorum()
         self.mine_quorum()
-        assert_sigs_nochange(True, False, True, 3)
+        assert_sigs_nochange(True, False, True, 3, other_ids=[first_id])
 
         # fast forward until 0.5 days before cleanup is expected, recovered sig should still be valid
         self.bump_mocktime(recsig_time + int(60 * 60 * 24 * 6.5) - self.mocktime, update_schedulers=False)
diff --git a/test/functional/feature_llmq_simplepose.py b/test/functional/feature_llmq_simplepose.py
index 6fa9e8f3f1cb..5bd61fe61723 100755
--- a/test/functional/feature_llmq_simplepose.py
+++ b/test/functional/feature_llmq_simplepose.py
@@ -179,6 +179,7 @@ def mine_quorum_no_service(self, mn, others, relay, services, expected_complaint
 
         self.log.info("Waiting for phase 2 (contribute)")
         self.wait_for_quorum_phase(q, 2, len(self.mninfo), "receivedContributions", self.llmq_size, self.mninfo, all_types=True)
+        self.wait_for_offset_quorum_phase(q, 2, self.mninfo)
 
         self.log.info("Replacing the connections to the masternode with stand-ins")
         mn.get_node(self).setnetworkactive(False)
@@ -194,21 +195,25 @@ def mine_quorum_no_service(self, mn, others, relay, services, expected_complaint
 
         self.log.info("Waiting for phase 3 (complain)")
         self.wait_for_quorum_phase(q, 3, len(others), "receivedComplaints", expected_complaints, others)
+        self.wait_for_offset_quorum_phase(q, 3, others, messages=False)
 
         self.move_blocks(nodes, 2)
 
         self.log.info("Waiting for phase 4 (justify)")
         self.wait_for_quorum_phase(q, 4, len(others), "receivedJustifications", 0, others)
+        self.wait_for_offset_quorum_phase(q, 4, others, messages=False)
 
         self.move_blocks(nodes, 2)
 
         self.log.info("Waiting for phase 5 (commit)")
         self.wait_for_quorum_phase(q, 5, len(others), "receivedPrematureCommitments", len(others), others)
+        self.wait_for_offset_quorum_phase(q, 5, others, messages=False)
 
         self.move_blocks(nodes, 2)
 
         self.log.info("Waiting for phase 6 (mining)")
         self.wait_for_quorum_phase(q, 6, len(others), None, 0, others)
+        self.wait_for_offset_quorum_phase(q, 6, others, messages=False)
 
         self.log.info("Waiting final commitment")
         self.wait_for_quorum_commitment(q, others)
@@ -225,6 +230,8 @@ def mine_quorum_no_service(self, mn, others, relay, services, expected_complaint
         self.wait_for_quorum_list(q, nodes)
         assert_equal(q, self.nodes[0].quorum("list", 1)["llmq_test"][0])
 
+        self.wait_for_offset_quorum_commitments(q, others)
+
         # Mine 8 (SIGN_HEIGHT_OFFSET) more blocks to make sure that the new quorum gets eligible for signing sessions
         self.generate(self.nodes[0], 8, sync_fun=lambda: self.sync_blocks(nodes))
 
@@ -263,22 +270,27 @@ def mine_quorum_less_checks(self, expected_members, mninfos_online):
 
         self.log.info("Waiting for phase 2 (contribute)")
         self.wait_for_quorum_phase(q, 2, expected_members, "receivedContributions", expected_members, mninfos_online, all_types=True)
+        self.wait_for_offset_quorum_phase(q, 2, mninfos_online)
         self.move_blocks(nodes, 2)
 
         self.log.info("Waiting for phase 3 (complain)")
         self.wait_for_quorum_phase(q, 3, expected_members, "receivedComplaints", None, mninfos_online, all_types=True)
+        self.wait_for_offset_quorum_phase(q, 3, mninfos_online)
         self.move_blocks(nodes, 2)
 
         self.log.info("Waiting for phase 4 (justify)")
         self.wait_for_quorum_phase(q, 4, expected_members, "receivedJustifications", None, mninfos_online, all_types=True)
+        self.wait_for_offset_quorum_phase(q, 4, mninfos_online)
         self.move_blocks(nodes, 2)
 
         self.log.info("Waiting for phase 5 (commit)")
         self.wait_for_quorum_phase(q, 5, expected_members, "receivedPrematureCommitments", expected_members, mninfos_online, all_types=True)
+        self.wait_for_offset_quorum_phase(q, 5, mninfos_online)
         self.move_blocks(nodes, 2)
 
         self.log.info("Waiting for phase 6 (mining)")
         self.wait_for_quorum_phase(q, 6, expected_members, None, 0, mninfos_online, all_types=True)
+        self.wait_for_offset_quorum_phase(q, 6, mninfos_online)
 
         self.log.info("Waiting final commitment")
         self.wait_for_quorum_commitment(q, mninfos_online)
@@ -298,6 +310,8 @@ def mine_quorum_less_checks(self, expected_members, mninfos_online):
         assert_equal(q, new_quorum)
         quorum_info = self.nodes[0].quorum("info", 100, new_quorum)
 
+        self.wait_for_offset_quorum_commitments(q, mninfos_online)
+
         # Mine 8 (SIGN_HEIGHT_OFFSET) more blocks to make sure that the new quorum gets eligible for signing sessions
         self.bump_mocktime(8)
         self.generate(self.nodes[0], 8, sync_fun=lambda: self.sync_blocks(nodes))
@@ -374,16 +388,16 @@ def test_banning(self, invalidate_proc, expected_connections=None):
 
     def repair_masternodes(self, restart):
         self.log.info("Repairing all banned and punished masternodes")
-        for mn in self.mninfo: # type: MasternodeInfo
-            if check_banned(self.nodes[0], mn) or check_punished(self.nodes[0], mn):
-                addr = self.nodes[0].getnewaddress()
-                self.nodes[0].sendtoaddress(addr, 0.1)
-                mn.update_service(self.nodes[0], submit=True, fundsAddr=addr)
-                if restart:
-                    self.stop_node(mn.nodeIdx)
-                    self.start_masternode(mn)
-                else:
-                    mn.get_node(self).setnetworkactive(True)
+        repaired = [mn for mn in self.mninfo if check_banned(self.nodes[0], mn) or check_punished(self.nodes[0], mn)]
+        for mn in repaired:
+            addr = self.nodes[0].getnewaddress()
+            self.nodes[0].sendtoaddress(addr, 0.1)
+            mn.update_service(self.nodes[0], submit=True, fundsAddr=addr)
+        if restart:
+            self.restart_masternodes(mns=repaired)
+        else:
+            for mn in repaired:
+                mn.get_node(self).setnetworkactive(True)
                 self.connect_nodes(mn.nodeIdx, 0)
 
         # syncing blocks only since node 0 has txes waiting to be mined
diff --git a/test/functional/p2p_blocksonly.py b/test/functional/p2p_blocksonly.py
index ebfdf87c393f..97a292258cec 100755
--- a/test/functional/p2p_blocksonly.py
+++ b/test/functional/p2p_blocksonly.py
@@ -71,6 +71,8 @@ def blocksonly_mode_tests(self):
             first_peer.send_message(msg_tx(tx))
             self.log.info('Check that the peer with relay-permission is still connected after sending the transaction')
             assert_equal(first_peer.is_connected, True)
+            # Let the node take the tx before mocktime moves, or its relay is scheduled past the bump
+            first_peer.sync_with_ping()
             self.bump_mocktime(60)
             second_peer.wait_for_tx(txid)
             assert_equal(self.nodes[0].getmempoolinfo()['size'], 1)
diff --git a/test/functional/test_framework/test_framework.py b/test/functional/test_framework/test_framework.py
index c7c722da7ccf..ee9bba3e7792 100755
--- a/test/functional/test_framework/test_framework.py
+++ b/test/functional/test_framework/test_framework.py
@@ -2080,6 +2080,10 @@ def wait_for_quorum_connections(self, quorum_hash, expected_connections, mninfos
             # wait_proc bumps mocktime by a second, keep it from outrunning the wall clock
             wait_proc = throttled(wait_proc)
 
+        # As in wait_for_quorum_phase: a PoSe-banned masternode can stay listed in a reused quarter of a
+        # rotating quorum, but it joins no DKG, so no connection to it is needed
+        online = {mn.proTxHash for mn in mninfos if not check_banned(self.nodes[0], mn)}
+
         def check_quorum_connections():
             def ret():
                 if wait_proc is not None:
@@ -2090,7 +2094,6 @@ def ret():
             # up as well, or its contributions are lost and wait_for_quorum_phase never sees them;
             # expected_connections only applies to the requested type
             found = False
-            online = {mn.proTxHash for mn in mninfos}
             for mn in mninfos:
                 s = mn.get_node(self).quorum("dkgstatus")
                 for qs in s["session"]:
@@ -2244,6 +2247,51 @@ def check_dkg_session():
 
         self.wait_until(check_dkg_session, timeout=timeout)
 
+    def offset_quorum_hash(self, quorum_hash):
+        """With rotation active, the index 1 session of a cycle is based on the block after quorum_hash and runs
+        one block behind the sessions of quorum_hash."""
+        return self.nodes[0].getblockhash(self.nodes[0].getblock(quorum_hash)["height"] + 1)
+
+    def wait_for_offset_quorum_phase(self, quorum_hash, phase, mninfos, messages=True):
+        """Called once the sessions of quorum_hash are in phase (2-6), when the offset session is still in
+        phase - 1 and the next phase block moves it on as well. A member whose worker has not caught up by then
+        aborts the session, so wait for every member of mninfos to be in phase - 1. With messages, also wait
+        until it is ready to move on: in phase 1 its relay connections must be up, because a contribution is
+        relayed only once and only to the members connected at that moment, and in phases 2-5 every member has
+        to hold every message of the phase that its members report having sent. Without messages (some member
+        is offline, so relays can be lost for good) only the phase is awaited. Passes at once without an offset
+        session."""
+        offset_hash = self.offset_quorum_hash(quorum_hash)
+        received = {2: "receivedContributions", 3: "receivedComplaints", 4: "receivedJustifications", 5: "receivedPrematureCommitments"}
+        check_received = received.get(phase - 1) if messages else None
+        self.wait_for_quorum_phase(offset_hash, phase - 1, 0, check_received, None, mninfos, all_types=True)
+        if phase == 2 and messages:
+            # Only the rotating llmq_test_dip0024 has a session based on the offset block
+            has_offset_session = any(qs["status"]["quorumHash"] == offset_hash
+                                     for mn in mninfos for qs in mn.get_node(self).quorum("dkgstatus")["session"])
+            if has_offset_session:
+                self.wait_for_quorum_connections(offset_hash, 0, mninfos, llmq_type_name="llmq_test_dip0024",
+                                                 wait_proc=lambda: self.bump_mocktime(1), all_types=True)
+
+    def wait_for_offset_quorum_commitments(self, quorum_hash, mninfos, timeout=15):
+        """The offset session finalizes with the block that mines the final commitments of quorum_hash, so wait
+        until no member's offset session is short of finalization and the commitments they produced reached
+        the mining node before mining past its window. A member whose session is done no longer reports it, and
+        an aborted session (e.g. with a member offline) produces no commitment. Passes at once without an offset
+        session."""
+        offset_hash = self.offset_quorum_hash(quorum_hash)
+
+        def check_finalized():
+            for mn in mninfos:
+                for qs in mn.get_node(self).quorum("dkgstatus")["session"]:
+                    status = qs["status"]
+                    if status["quorumHash"] == offset_hash and not status["aborted"] and status["phase"] < 6:
+                        return False
+            return True
+
+        self.wait_until(check_finalized, timeout=timeout)
+        self.wait_for_quorum_commitments_on_miner(offset_hash, mninfos)
+
     def node_quorum_commitment_types(self, node, quorum_hash):
         s = node.quorum("dkgstatus")
         return {c["llmqType"] for c in s.get("minableCommitments", [])
@@ -2362,26 +2410,31 @@ def mine_quorum(self, llmq_type_name="llmq_test", llmq_type=100, expected_connec
 
         self.log.info("Waiting for phase 2 (contribute)")
         self.wait_for_quorum_phase(q, 2, expected_members, "receivedContributions", expected_contributions, mninfos_online, llmq_type_name=llmq_type_name, all_types=all_types)
+        self.wait_for_offset_quorum_phase(q, 2, mninfos_online, messages=all_types)
 
         self.move_blocks(nodes, 2)
 
         self.log.info("Waiting for phase 3 (complain)")
         self.wait_for_quorum_phase(q, 3, expected_members, "receivedComplaints", expected_complaints, mninfos_online, llmq_type_name=llmq_type_name, all_types=all_types)
+        self.wait_for_offset_quorum_phase(q, 3, mninfos_online, messages=all_types)
 
         self.move_blocks(nodes, 2)
 
         self.log.info("Waiting for phase 4 (justify)")
         self.wait_for_quorum_phase(q, 4, expected_members, "receivedJustifications", expected_justifications, mninfos_online, llmq_type_name=llmq_type_name, all_types=all_types)
+        self.wait_for_offset_quorum_phase(q, 4, mninfos_online, messages=all_types)
 
         self.move_blocks(nodes, 2)
 
         self.log.info("Waiting for phase 5 (commit)")
         self.wait_for_quorum_phase(q, 5, expected_members, "receivedPrematureCommitments", expected_commitments, mninfos_online, llmq_type_name=llmq_type_name, all_types=all_types)
+        self.wait_for_offset_quorum_phase(q, 5, mninfos_online, messages=all_types)
 
         self.move_blocks(nodes, 2)
 
         self.log.info("Waiting for phase 6 (mining)")
         self.wait_for_quorum_phase(q, 6, expected_members, None, 0, mninfos_online, llmq_type_name=llmq_type_name, all_types=all_types)
+        self.wait_for_offset_quorum_phase(q, 6, mninfos_online, messages=all_types)
 
         self.log.info("Waiting final commitment")
         self.wait_for_quorum_commitment(q, mninfos_online, llmq_type=llmq_type)
@@ -2404,6 +2457,7 @@ def mine_quorum(self, llmq_type_name="llmq_test", llmq_type=100, expected_connec
             self.verify_upcoming_dkg_predictions(predictions, quorum_info)
 
         if not skip_maturity:
+            self.wait_for_offset_quorum_commitments(q, mninfos_online)
             # Mine 8 (SIGN_HEIGHT_OFFSET) more blocks to make sure that the new quorum gets eligible for signing sessions
             self.generate(self.nodes[0], 8, sync_fun=lambda: self.sync_blocks(nodes))