test: add coverage for assumeUTXO honest peers disconnection

Exercising and verifying the following points:

1. An IBD node can sync headers from an AssumeUTXO node at
   any time.

2. IBD nodes do not request historical blocks from AssumeUTXO
   nodes while they are syncing the background-chain.

3. The assumeUTXO node dynamically adjusts the network services
   it offers according to its state.

4. IBD nodes can fully sync from AssumeUTXO nodes after they
   finish the background-chain sync.

Github-Pull: bitcoin#30807
Rebased-From: 992f83b

test/functional/feature_assumeutxo.py

@@ -9,19 +9,29 @@
 The assumeutxo value generated and used here is committed to in
 `CRegTestParams::m_assumeutxo_data` in `src/kernel/chainparams.cpp`.
 """
+import time
 from shutil import rmtree
 
 from dataclasses import dataclass
 from test_framework.blocktools import (
         create_block,
         create_coinbase
 )
-from test_framework.messages import tx_from_hex
+from test_framework.messages import (
+    CBlockHeader,
+    from_hex,
+    msg_headers,
+    tx_from_hex
+)
+from test_framework.p2p import (
+    P2PInterface,
+)
 from test_framework.test_framework import BitcoinTestFramework
 from test_framework.util import (
     assert_approx,
     assert_equal,
     assert_raises_rpc_error,
+    try_rpc,
 )
 from test_framework.wallet import (
     getnewdestination,
@@ -247,6 +257,69 @@ def test_snapshot_not_on_most_work_chain(self, dump_output_path):
         node1.submitheader(main_block1)
         node1.submitheader(main_block2)
 
+    def test_sync_from_assumeutxo_node(self, snapshot):
+        """
+        This test verifies that:
+        1. An IBD node can sync headers from an AssumeUTXO node at any time.
+        2. IBD nodes do not request historical blocks from AssumeUTXO nodes while they are syncing the background-chain.
+        3. The assumeUTXO node dynamically adjusts the network services it offers according to its state.
+        4. IBD nodes can fully sync from AssumeUTXO nodes after they finish the background-chain sync.
+        """
+        self.log.info("Testing IBD-sync from assumeUTXO node")
+        # Node2 starts clean and loads the snapshot.
+        # Node3 starts clean and seeks to sync-up from snapshot_node.
+        miner = self.nodes[0]
+        snapshot_node = self.nodes[2]
+        ibd_node = self.nodes[3]
+
+        # Start test fresh by cleaning up node directories
+        for node in (snapshot_node, ibd_node):
+            self.stop_node(node.index)
+            rmtree(node.chain_path)
+            self.start_node(node.index, extra_args=self.extra_args[node.index])
+
+        # Sync-up headers chain on snapshot_node to load snapshot
+        headers_provider_conn = snapshot_node.add_p2p_connection(P2PInterface())
+        headers_provider_conn.wait_for_getheaders()
+        msg = msg_headers()
+        for block_num in range(1, miner.getblockcount()+1):
+            msg.headers.append(from_hex(CBlockHeader(), miner.getblockheader(miner.getblockhash(block_num), verbose=False)))
+        headers_provider_conn.send_message(msg)
+
+        # Ensure headers arrived
+        default_value = {'status': ''}  # No status
+        headers_tip_hash = miner.getbestblockhash()
+        self.wait_until(lambda: next(filter(lambda x: x['hash'] == headers_tip_hash, snapshot_node.getchaintips()), default_value)['status'] == "headers-only")
+        snapshot_node.disconnect_p2ps()
+
+        # Load snapshot
+        snapshot_node.loadtxoutset(snapshot['path'])
+
+        # Connect nodes and verify the ibd_node can sync-up the headers-chain from the snapshot_node
+        self.connect_nodes(ibd_node.index, snapshot_node.index)
+        snapshot_block_hash = snapshot['base_hash']
+        self.wait_until(lambda: next(filter(lambda x: x['hash'] == snapshot_block_hash, ibd_node.getchaintips()), default_value)['status'] == "headers-only")
+
+        # Once the headers-chain is synced, the ibd_node must avoid requesting historical blocks from the snapshot_node.
+        # If it does request such blocks, the snapshot_node will ignore requests it cannot fulfill, causing the ibd_node
+        # to stall. This stall could last for up to 10 min, ultimately resulting in an abrupt disconnection due to the
+        # ibd_node's perceived unresponsiveness.
+        time.sleep(3)  # Sleep here because we can't detect when a node avoids requesting blocks from other peer.
+        assert_equal(len(ibd_node.getpeerinfo()[0]['inflight']), 0)
+
+        # Now disconnect nodes and finish background chain sync
+        self.disconnect_nodes(ibd_node.index, snapshot_node.index)
+        self.connect_nodes(snapshot_node.index, miner.index)
+        self.sync_blocks(nodes=(miner, snapshot_node))
+        # Check the base snapshot block was stored and ensure node signals full-node service support
+        self.wait_until(lambda: not try_rpc(-1, "Block not found", snapshot_node.getblock, snapshot_block_hash))
+        assert 'NETWORK' in snapshot_node.getnetworkinfo()['localservicesnames']
+
+        # Now the snapshot_node is sync, verify the ibd_node can sync from it
+        self.connect_nodes(snapshot_node.index, ibd_node.index)
+        assert 'NETWORK' in ibd_node.getpeerinfo()[0]['servicesnames']
+        self.sync_blocks(nodes=(ibd_node, snapshot_node))
+
     def assert_only_network_limited_service(self, node):
         node_services = node.getnetworkinfo()['localservicesnames']
         assert 'NETWORK' not in node_services
@@ -623,6 +696,9 @@ def check_tx_counts(final: bool) -> None:
 
         self.test_snapshot_in_a_divergent_chain(dump_output['path'])
 
+        # The following test cleans node2 and node3 chain directories.
+        self.test_sync_from_assumeutxo_node(snapshot=dump_output)
+
 @dataclass
 class Block:
     hash: str