double_tx_relay: use miniwallet

Author: willcl-ark

src/scenarios/double_tx_relay.py

@@ -1,17 +1,13 @@
 #!/usr/bin/env python3
-import asyncio
-import random
-import math
 import time
 from datetime import datetime
 from decimal import Decimal
-
-from warnet.test_framework_bridge import WarnetTestFramework
-from scenarios.utils import ensure_miner, next_block_delta, get_block_reward_sats
-from test_framework.test_node import TestNode
-import concurrent.futures
 from threading import Lock
 
+from test_framework.authproxy import JSONRPCException
+from test_framework.blocktools import COINBASE_MATURITY
+from test_framework.wallet import MiniWallet
+from warnet.test_framework_bridge import WarnetTestFramework
 
 BLOCKS_WAIT_TILL_SPENDABLE = 101
 MIN_UTXO_AMOUNT = Decimal(0.001)
@@ -31,48 +27,9 @@ def add_options(self, parser):
             "--txproductionrate",
             dest="txproductionrate",
             default=5,
-            # action="store_true",
+            type=int,
             help="Rate per second at which transactions are generated. (default 5 tx/s)",
         )
-        parser.add_argument(
-            "--speedyinitialblockrewards",
-            dest="speedyinitialblockrewards",
-            default=True,
-            type=bool,
-            help="Mines blocks quickly at first to get the first rewards available. (default True)",
-        )
-
-    def enqueue_block(self, block_and_node):
-        self.block_queue.append(block_and_node)
-
-    def dequeue_spendable_block_if_available(self):
-        if len(self.block_queue) < BLOCKS_WAIT_TILL_SPENDABLE:
-            return None
-        return self.block_queue.pop(0)
-
-    def get_node_ip(self, node):
-        # loop 3 times to get the ip address
-        for n in range(3):
-            ipaddress = node.getnetworkinfo()["localaddresses"][n]["address"]
-            if ipaddress != "0.0.0.0" and ipaddress != "::":
-                return ipaddress
-        return None
-
-    def connect_nodes(self):
-        nodes_without_peers = []
-        for node in self.nodes:
-            try:
-                if node.getpeerinfo() == []:
-                    nodes_without_peers.append(node.index)
-                    random_node = random.choice(self.nodes)
-                    ipaddress = self.get_node_ip(random_node)
-                    node.addnode(f"{ipaddress}:18444", "add")
-            except Exception:
-                pass
-        if len(nodes_without_peers) == 0:
-            return False
-        self.log.info("Nodes without peers: %s", nodes_without_peers)
-        return True
 
     def run_test(self):
         self.log.info("Starting Double TX Relay Scenario")
@@ -81,160 +38,64 @@ def run_test(self):
         self.sent_txs = 0  # protected by mutex
         self.time_of_last_log = datetime.now()  # protected by mutex
         self.failed_txs = 0  # protected by mutex
-
-        self.log.info("Checking all nodes have a peer")
-        # ensure all nodes have a peer
-        while self.connect_nodes():
-            self.log.info("Waiting for all nodes to have a peer")
-            time.sleep(5)
-
-        if self.options.speedyinitialblockrewards:
-            self.log.info("Generating initial blocks and rewards")
-            for node in self.nodes:
-                self.generate_block(self.nodes[0], node)
-            for _ in range(BLOCKS_WAIT_TILL_SPENDABLE):
-                self.generate_block(self.nodes[0], self.nodes[0])
-
-        self.log.info("Starting block mining and tx sending in real time")
-
-        asyncio.run(self.mainasync())
-
-    def generate_block(self, generating_node: TestNode, receiving_node: TestNode):
-        try:
-            self.log.info("Generating block for node: %s", receiving_node.index)
-            wallet = ensure_miner(receiving_node)
-            addr = wallet.getnewaddress(address_type="bech32m")
-            block = self.generatetoaddress(generating_node, 1, addr)[0]
-            # allow time for block to propagate
-            time.sleep(3)
-            self.enqueue_block((block, receiving_node))
-            spendable_block_and_node = self.dequeue_spendable_block_if_available()
-            if spendable_block_and_node is not None:
-                self.split_block_up(spendable_block_and_node)
-        except Exception as e:
-            self.log.error("Error generating block: %s", e)
-
-    def split_block_up(self, block_and_node):
-        block = block_and_node[0]
-        node = block_and_node[1]
-        try:
-            # TODO: check if block is orphaned?
-            height = node.getblock(block)["height"]
-            reward = get_block_reward_sats(height)
-            self.log.info(
-                "Splitting %s sat block reward at height %s reward up for node: %s",
-                reward,
-                height,
-                node.index,
-            )
-            wallet = ensure_miner(node)
-            addresses = {}
-            for _ in range(math.floor(reward / 10000000) - 1):
-                addresses[wallet.getnewaddress(address_type="bech32m")] = "0.1"
-            node.send(outputs=addresses)
-        except Exception as e:
-            self.log.error("Error splitting block up: %s", e)
-
-    async def mainasync(self):
-        await asyncio.gather(self.mineblocks(), self.sendtransactions())
-
-    async def mineblocks(self):
-        print(
-            f"Starting process to select a random node to mine a block roughly every {AVERAGE_BLOCK_TIME} seconds"
-        )
-        while True:
-            try:
-                # trunk-ignore(bandit/B311)
-                node = random.choice(self.nodes)
-                self.generate_block(node, node)
-                delay = next_block_delta(AVERAGE_BLOCK_TIME)
-                self.log.info(f"Waiting {'%.0f' % delay} seconds to mine next block")
-                await asyncio.sleep(delay)
-            except Exception as e:
-                self.log.error("Error mining blocks: %s", e)
-                await asyncio.sleep(10)
-
-    async def sendtransactions(self):
-        print(
+        self.node_details = {}
+        self.miniwallet = MiniWallet(self.nodes[0])
+        num_blocks = COINBASE_MATURITY + 100
+        self.log.info(f"Generating {num_blocks} blocks...")
+        self.generate(self.miniwallet, num_blocks)
+        self.log.info("Rescanning utxos")
+        self.miniwallet.rescan_utxos()
+
+        split_num = 1000
+        confirmed_utxos = self.miniwallet.get_utxos(confirmed_only=True)
+        print(f"Starting with {len(confirmed_utxos)} confirmed utxos, now splitting each into 1000...")
+        for i, utxo in enumerate(confirmed_utxos):
+            tx = self.miniwallet.send_self_transfer_multi(from_node=self.nodes[0], utxos_to_spend=[utxo], num_outputs=split_num)
+            self.log.info(f"Sent tx {i} of {len(confirmed_utxos)}: {tx['txid']}")
+        self.generate(self.miniwallet, 6)
+        self.miniwallet.rescan_utxos()
+
+        confirmed_utxos = self.miniwallet.get_utxos(confirmed_only=True)
+        self.log.info(f"Now got {len(confirmed_utxos)} utxos")
+
+        self.log.info(
             f"Starting process to send transactions at a rate of {self.options.txproductionrate} tx/s"
         )
-        with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
-            while True:
-                executor.submit(self.send_transaction)
-                # await asyncio.sleep(10)
-                await asyncio.sleep(1 / int(self.options.txproductionrate))
-
-    def send_transaction(self):
-        try:
-            # trunk-ignore(bandit/B311)
-            from_node = random.choice(self.nodes)
-            # trunk-ignore(bandit/B311)
-            to_node = random.choice(self.nodes)
-            wallet = ensure_miner(from_node)
-            utxos = wallet.listunspent(include_unsafe=False, query_options={"maximumAmount": 0.1})
-            if len(utxos) == 0:
-                return
-            utxo = random.choice(utxos)
-            fee_rate = Decimal("0.00001000")
-            try:
-                multiplier = Decimal(math.exp(random.random() * 0.2 - 0.1))
-                fee_rate = from_node.estimatesmartfee(5)["feerate"] * multiplier
-                fee_rate = fee_rate.quantize(Decimal(".00000001"))
-
-            # trunk-ignore(bandit/B110)
-            except Exception:
-                pass
-            # 1 in 1 out taproot transaction should be 111 vbytes
-            fee = fee_rate * Decimal("0.111")
-            if fee < Decimal("0.00000111"):
-                fee = Decimal("0.00000111")
-            amount = Decimal(utxo["amount"])
-            amount = amount - fee
-            amount = amount.quantize(Decimal(".00000001"))
-
-            raw_transaction_inputs = [
-                {
-                    "txid": utxo["txid"],
-                    "vout": utxo["vout"],
-                },
-            ]
-            raw_transaction_outputs = [
-                {
-                    to_node.getnewaddress(address_type="bech32m"): amount,
-                },
-            ]
-            if amount < Decimal(MIN_UTXO_AMOUNT):
-                raw_transaction_outputs = [
-                    {
-                        "data": "",
-                    },
-                ]
-            tx = from_node.createrawtransaction(
-                inputs=raw_transaction_inputs, outputs=raw_transaction_outputs
-            )
-            tx = from_node.signrawtransactionwithwallet(tx)["hex"]
-            from_node.sendrawtransaction(tx)
-            # runs in thread pool so mutex is needed
-            with self.mutex:
-                self.sent_txs += 1
-                seconds_since_log = (datetime.now() - self.time_of_last_log).total_seconds()
-                if seconds_since_log > 59:
-                    self.log.info(
-                        f"Sent roughly {'%.2f' % (self.sent_txs / 60)} transactions per second over the last minute"
-                    )
-                    self.log.info(
-                        f"Roughly {'%.2f' % (self.failed_txs / 60)} transactions per second have failed over the last minute"
-                    )
-                    self.sent_txs = 0
-                    self.failed_txs = 0
-                    self.time_of_last_log = datetime.now()
-        except Exception as e:
-            with self.mutex:
-                self.failed_txs += 1
-            self.log.error("Error sending transaction: %s", e)
-
-
-# bcli -rpcwallet=test-wallet -named send outputs="{\"$ADDRESS\": 1}" fee_rate=10
+        tx_count = 0
+        start_time = time.time()
+        interval = 1 / self.options.txproductionrate
+        time_to_next_tx = 0
+        next_log_time = start_time + 10
+
+        while True:
+            for node in self.nodes:
+                time.sleep(time_to_next_tx)
+
+                tx_start_time = time.time()
+                if len(confirmed_utxos) == 0:
+                    self.generate(self.miniwallet, 1)
+                    self.miniwallet.rescan_utxos()
+                    confirmed_utxos = self.miniwallet.get_utxos(confirmed_only=True)
+                utxo = confirmed_utxos.pop()
+                try:
+                    self.miniwallet.send_self_transfer(from_node=node, utxo_to_spend=utxo)
+                    tx_count += 1
+                except JSONRPCException as e:
+                    self.log.warning(f"tx failed: {e}, continuing")
+
+                # Adjust next sleep
+                tx_end_time = time.time()
+                tx_duration = tx_end_time - tx_start_time
+                time_to_next_tx = max(0, interval - tx_duration)
+
+                current_time = time.time()
+                elapsed_time = current_time - start_time
+
+                if current_time >= next_log_time:
+                    tps = tx_count / elapsed_time
+                    self.log.info(f"Transactions per second (TPS): {tps}")
+                    next_log_time += 10
+
 
 if __name__ == "__main__":
     print("Running Double TX Relay Scenario")