rostrum 14.0.1

An efficient implementation of Electrum Server with token support
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

#!/usr/bin/env python3
# Copyright (c) 2026 The Bitcoin Unlimited developers

"""
Test that reproduces the faster_statushash rscan ordering bug.

The bug: faster_statushash uses rscan (reverse DB scan) to cap entries at
max_elements. The DB key is (scripthash, flags, outpointhash) with no height
component. So rscan returns entries by descending outpointhash (random hash
order), NOT by most-recent block height. When total entries exceed
max_elements, newly confirmed entries may fall outside the rscan window,
causing statushash to remain unchanged even though the address history changed.

This test sets ROSTRUM_STATUSHASH_MAX_ELEMENTS to a small value, creates many
more entries than that, then asserts that each new confirmed transaction
changes the statushash.
"""

import os

# Must be set before the test framework imports and starts processes.
os.environ["ROSTRUM_STATUSHASH_MAX_ELEMENTS"] = "5"

# pylint: disable=wrong-import-position
import asyncio
from test_framework.electrumutil import ElectrumTestFramework
from test_framework.electrumconnection import ElectrumConnection
from test_framework.environment import on_bch, on_nex
from test_framework.utiltx import p2pkh

if on_bch():
    from test_framework.bch.blocktools import create_transaction
    from test_framework.bch.script import anyonecanspend_unlocking_script
elif on_nex():
    from test_framework.nex.blocktools import create_transaction
    from test_framework.nex.script import anyonecanspend_unlocking_script
    from test_framework.nex.nodemessages import TxOut
else:
    raise NotImplementedError()

SCRIPTHASH_SUBSCRIBE = "blockchain.scripthash.subscribe"

NUM_INITIAL_TXS = 50
NUM_ITERATIONS = 3


async def address_to_scripthash(address):
    cli = ElectrumConnection()
    await cli.connect()
    scripthash = await cli.call("blockchain.address.get_scripthash", address)
    cli.disconnect()
    return scripthash


def make_conform_to_ctor(txs):
    for tx in txs:
        tx.rehash()
    return sorted(txs, key=lambda tx: tx.get_rpc_hex_id())


class StatushashBugTest(ElectrumTestFramework):
    """
    Reproduces the faster_statushash bug where new confirmed transactions
    don't change the statushash when entries exceed max_elements.
    """

    async def run_test(self):
        n = self.nodes[0]
        await self.bootstrap_p2p()

        cli = ElectrumConnection()
        await cli.connect()
        try:
            coinbases = await self.mine_blocks(cli, n, 201)

            addr = n.getnewaddress("p2pkh")
            scripthash = await address_to_scripthash(addr)

            self.info(
                "Creating %d transactions to address to exceed max_elements (5)",
                NUM_INITIAL_TXS,
            )
            txs = []
            out_type = (
                TxOut.TYPE_SATOSCRIPT  # pylint: disable=possibly-used-before-assignment
                if on_nex()
                else 0
            )
            for _ in range(NUM_INITIAL_TXS):
                coinbase = coinbases.pop(0)
                tx = create_transaction(
                    prevtx=coinbase,
                    value=10000,
                    n=0,
                    sig=anyonecanspend_unlocking_script(),
                    out=p2pkh(addr),
                    out_type=out_type,
                )
                txs.append(tx)

            txs = make_conform_to_ctor(txs)
            await self.mine_blocks(cli, n, 1, txs)

            self.info("Subscribing to scripthash to get initial statushash")
            old_statushash, _ = await cli.subscribe(SCRIPTHASH_SUBSCRIBE, scripthash)
            assert (
                old_statushash is not None
            ), "Address with history should have a statushash"
            self.info("Initial statushash: %s", old_statushash)

            for i in range(NUM_ITERATIONS):
                self.info("Iteration %d: mining 1 new tx to the address", i + 1)
                coinbase = coinbases.pop(0)
                tx = create_transaction(
                    prevtx=coinbase,
                    value=10000,
                    n=0,
                    sig=anyonecanspend_unlocking_script(),
                    out=p2pkh(addr),
                    out_type=out_type,
                )
                tx.rehash()
                await self.mine_blocks(cli, n, 1, [tx])

                self.info("Re-subscribing to get updated statushash")
                new_statushash, _ = await cli.subscribe(
                    SCRIPTHASH_SUBSCRIBE, scripthash
                )

                assert new_statushash is not None, "Statushash should not be None"
                assert (
                    new_statushash != old_statushash
                ), f"Iteration {i + 1}: statushash should change after new confirmed tx (old={old_statushash}, new={new_statushash})"

                self.info(
                    "Iteration %d: statushash changed %s -> %s",
                    i + 1,
                    old_statushash,
                    new_statushash,
                )
                old_statushash = new_statushash

            self.info("All iterations passed - statushash updated correctly")
        finally:
            cli.disconnect()


if __name__ == "__main__":
    asyncio.run(StatushashBugTest().main())