fork_parent_height_test.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2026 The Limenka developers
3 # Distributed under the MIT software license, see the accompanying
4 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
5 """Fork node stops syncing from a spamchain parent peer at the fork point.
6
7 Once the hard-coded parent-chain fork-point height (nForkActivationHeight)
8 is reached, the parent chain is no longer authoritative: parent peers stop
9 serving blocks (they can no longer satisfy fork consensus past that height)
10 and the fork node stops downloading from them. The parent connection is
11 kept alive for the shared mempool.
12
13 Node A impersonates a spamchain node (speaks mainnet magic) and mines a
14 mainnet-style chain the whole time. Node B has -forkactivationheight=50,
15 so it accepts A's blocks up to height 50 as pre-fork history and must stop
16 there - even though A keeps mining to height 100.
17 """
18 import hashlib
19
20 from test_framework.address import byte_to_base58, hash160
21 from test_framework.descriptors import descsum_create
22 from test_framework.key import ECKey
23 from test_framework.segwit_addr import encode_segwit_address
24 from test_framework.test_framework import LimenkaTestFramework
25 from test_framework.util import p2p_port
26
27 GENESIS_TIME = 1231006505
28 ACTIVATION_FAR = 1999999999
29 MAINNET_MAGIC = 'f9beb4d9'
30 FORK_HEIGHT = 50
31
32
33 class ForkParentHeightTest(LimenkaTestFramework):
34 def set_test_params(self):
35 self.setup_clean_chain = True
36 self.num_nodes = 2
37 self.chain = 'limenka'
38 self.extra_args = [[
39 '-forkactivationtime=%d' % ACTIVATION_FAR,
40 '-forkdelaysteps=1024',
41 '-forkmineondemand',
42 '-forkstandalone',
43 '-fallbackfee=0.0002',
44 '-addresstype=bech32',
45 # A impersonates a spamchain node (speaks mainnet magic).
46 f'-p2pmagic={MAINNET_MAGIC}',
47 ], [
48 # B's fork is permanently active at height FORK_HEIGHT.
49 '-forkactivationtime=%d' % ACTIVATION_FAR,
50 '-forkactivationheight=%d' % FORK_HEIGHT,
51 '-forkdelaysteps=1024',
52 '-forkmineondemand',
53 '-forkstandalone',
54 '-fallbackfee=0.0002',
55 '-addresstype=bech32',
56 f'-parentpeer=127.0.0.1:{p2p_port(0)}',
57 ]]
58 self.rpc_timeout = 240
59
60 def skip_test_if_missing_module(self):
61 self.skip_if_no_wallet()
62
63 def add_options(self, parser):
64 self.add_wallet_options(parser, descriptors=True, legacy=False)
65
66 def init_wallet(self, *, node):
67 wallet_name = self.default_wallet_name if self.wallet_names is None else self.wallet_names[node] if node < len(self.wallet_names) else False
68 if wallet_name is not False:
69 n = self.nodes[node]
70 if wallet_name is not None and node == 0:
71 n.createwallet(wallet_name=wallet_name, descriptors=True, load_on_startup=True)
72 seed = hashlib.sha256(b'fork-parent-height-funding').digest()
73 key = ECKey()
74 key.set(seed, compressed=True)
75 priv = key.get_bytes()
76 wif = byte_to_base58(priv + b'\x01', 128)
77 desc = descsum_create(f"wpkh({wif})")
78 import_res = n.importdescriptors([{
79 "desc": desc,
80 "timestamp": GENESIS_TIME + 600,
81 "active": False,
82 "internal": False,
83 }])
84 assert all(r['success'] for r in import_res)
85 self.funding_pub = key.get_pubkey().get_bytes()
86
87 def setup_network(self):
88 self.setup_nodes()
89
90 def run_test(self):
91 node_a, node_b = self.nodes[0], self.nodes[1]
92
93 p2wpkh_program = hash160(self.funding_pub)
94 addr = encode_segwit_address("bf", 0, p2wpkh_program)
95 self.generatetoaddress(node_a, 100, addr, sync_fun=self.no_op)
96 assert node_a.getblockcount() == 100
97
98 # B must stop exactly at the fork-point height, never past it.
99 self.wait_until(
100 lambda: node_b.getblockcount() == FORK_HEIGHT,
101 timeout=240,
102 )
103 info_b = node_b.getblockchaininfo()
104 assert info_b['blocks'] == FORK_HEIGHT, info_b
105 assert info_b['headers'] >= FORK_HEIGHT, info_b
106
107 # The pre-fork chain matches A exactly up to the fork point.
108 assert node_b.getblockhash(FORK_HEIGHT) == node_a.getblockhash(FORK_HEIGHT)
109
110 # B's chain hash at the fork point is a pre-fork (parent-rule) block.
111 assert info_b['bestblockhash'] == node_a.getblockhash(FORK_HEIGHT)
112
113 # The parent connection stays up (kept for the shared mempool).
114 peers = node_b.getpeerinfo()
115 assert any(p['addr'] == f"127.0.0.1:{p2p_port(0)}" for p in peers), peers
116
117 self.log.info("parent sync disengaged at fork height %d (A at %d)" % (FORK_HEIGHT, node_a.getblockcount()))
118
119
120 if __name__ == '__main__':
121 ForkParentHeightTest(__file__).main()
122