mempool_subdust_fee_penalty.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2026 The Limenka Knots developers
3 # Distributed under the MIT software license, see the accompanying
4 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
5 """Test sub-dust output fee penalty (-subdustfeepenalty)."""
6
7 from math import ceil
8
9 from test_framework.messages import (
10 COIN,
11 COutPoint,
12 CTransaction,
13 CTxIn,
14 CTxOut,
15 )
16 from test_framework.script import (
17 CScript,
18 OP_RETURN,
19 )
20 from test_framework.test_framework import LimenkaTestFramework
21 from test_framework.util import assert_equal
22 from test_framework.wallet import MiniWallet
23
24
25 DUST_THRESHOLD = 330 # P2TR: (43 + 67) * 3000 / 1000 = 330 sats
26
27
28 class SubDustFeePenaltyTest(LimenkaTestFramework):
29 def set_test_params(self):
30 self.num_nodes = 1
31 self.extra_args = [["-acceptnonstdtxn=1", "-subdustfeepenalty=1"]]
32
33 def run_test(self):
34 self.wallet = MiniWallet(self.nodes[0])
35
36 info = self.nodes[0].getmempoolinfo()
37 self.minrelay_per_kvb = info['minrelaytxfee'] * COIN # sats per kvB
38
39 self.test_dust_output_increases_required_fee()
40 self.test_partial_dust_proportional_penalty()
41 self.test_multiple_dust_outputs_stack()
42 self.test_penalty_disabled()
43 self.test_op_return_not_penalized()
44 self.test_above_dust_not_penalized()
45
46 def build_tx_with_dust(self, dust_values, fee):
47 utxo = self.wallet.get_utxo()
48 tx = CTransaction()
49 tx.version = 2
50 tx.vin = [CTxIn(COutPoint(int(utxo["txid"], 16), utxo["vout"]))]
51 input_value = int(utxo["value"] * COIN)
52 change = input_value - sum(dust_values) - fee
53 assert change > 0, f"Not enough funds: input={input_value}, dust={dust_values}, fee={fee}"
54 tx.vout = [CTxOut(v, self.wallet.get_output_script()) for v in dust_values]
55 tx.vout.append(CTxOut(change, self.wallet.get_output_script()))
56 self.wallet.sign_tx(tx)
57 return tx
58
59 def get_min_relay_fee(self, tx):
60 """Calculate the minimum relay fee for a transaction."""
61 decoded = self.nodes[0].decoderawtransaction(tx.serialize().hex())
62 return int(ceil(self.minrelay_per_kvb * decoded['vsize'] / 1000))
63
64 def test_dust_output_increases_required_fee(self):
65 self.log.info("Test: sub-dust output penalizes effective fee")
66
67 # Build a probe tx to determine the exact min relay fee
68 probe = self.build_tx_with_dust([0], fee=1000)
69 min_fee = self.get_min_relay_fee(probe)
70
71 # fee just covers min relay + penalty - 1: rejected
72 reject_fee = DUST_THRESHOLD + min_fee - 1
73 tx_low = self.build_tx_with_dust([0], fee=reject_fee)
74 result = self.nodes[0].testmempoolaccept([tx_low.serialize().hex()])
75 assert_equal(result[0]["allowed"], False)
76 assert_equal(result[0]["reject-reason"], "min relay fee not met")
77
78 # fee covers min relay + penalty: accepted
79 accept_fee = DUST_THRESHOLD + min_fee
80 tx_high = self.build_tx_with_dust([0], fee=accept_fee)
81 result = self.nodes[0].testmempoolaccept([tx_high.serialize().hex()])
82 assert_equal(result[0]["allowed"], True)
83
84 def test_partial_dust_proportional_penalty(self):
85 self.log.info("Test: partial dust value gets proportional penalty")
86
87 partial_value = 100
88 penalty = DUST_THRESHOLD - partial_value # 230 sats
89
90 probe = self.build_tx_with_dust([partial_value], fee=1000)
91 min_fee = self.get_min_relay_fee(probe)
92
93 reject_fee = penalty + min_fee - 1
94 tx_low = self.build_tx_with_dust([partial_value], fee=reject_fee)
95 result = self.nodes[0].testmempoolaccept([tx_low.serialize().hex()])
96 assert_equal(result[0]["allowed"], False)
97 assert_equal(result[0]["reject-reason"], "min relay fee not met")
98
99 accept_fee = penalty + min_fee
100 tx_high = self.build_tx_with_dust([partial_value], fee=accept_fee)
101 result = self.nodes[0].testmempoolaccept([tx_high.serialize().hex()])
102 assert_equal(result[0]["allowed"], True)
103
104 def test_multiple_dust_outputs_stack(self):
105 self.log.info("Test: multiple dust outputs stack penalties")
106
107 total_penalty = DUST_THRESHOLD * 2 # 660 sats
108
109 probe = self.build_tx_with_dust([0, 0], fee=1000)
110 min_fee = self.get_min_relay_fee(probe)
111
112 reject_fee = total_penalty + min_fee - 1
113 tx_low = self.build_tx_with_dust([0, 0], fee=reject_fee)
114 result = self.nodes[0].testmempoolaccept([tx_low.serialize().hex()])
115 assert_equal(result[0]["allowed"], False)
116 assert_equal(result[0]["reject-reason"], "min relay fee not met")
117
118 accept_fee = total_penalty + min_fee
119 tx_high = self.build_tx_with_dust([0, 0], fee=accept_fee)
120 result = self.nodes[0].testmempoolaccept([tx_high.serialize().hex()])
121 assert_equal(result[0]["allowed"], True)
122
123 def test_penalty_disabled(self):
124 self.log.info("Test: penalty disabled with -subdustfeepenalty=0")
125 self.restart_node(0, extra_args=["-acceptnonstdtxn=1", "-subdustfeepenalty=0"])
126 self.wallet.rescan_utxos()
127
128 probe = self.build_tx_with_dust([0], fee=1000)
129 min_fee = self.get_min_relay_fee(probe)
130
131 # Without penalty, just the min relay fee suffices
132 tx = self.build_tx_with_dust([0], fee=min_fee)
133 result = self.nodes[0].testmempoolaccept([tx.serialize().hex()])
134 assert_equal(result[0]["allowed"], True)
135
136 def test_op_return_not_penalized(self):
137 self.log.info("Test: OP_RETURN outputs are not penalized (threshold=0)")
138 self.restart_node(0, extra_args=["-acceptnonstdtxn=1", "-subdustfeepenalty=1"])
139 self.wallet.rescan_utxos()
140
141 utxo = self.wallet.get_utxo()
142 tx = CTransaction()
143 tx.version = 2
144 tx.vin = [CTxIn(COutPoint(int(utxo["txid"], 16), utxo["vout"]))]
145 input_value = int(utxo["value"] * COIN)
146 fee = 400
147 tx.vout = [
148 CTxOut(0, CScript([OP_RETURN, b"test data"])),
149 CTxOut(input_value - fee, self.wallet.get_output_script()),
150 ]
151 self.wallet.sign_tx(tx)
152
153 result = self.nodes[0].testmempoolaccept([tx.serialize().hex()])
154 assert_equal(result[0]["allowed"], True)
155
156 def test_above_dust_not_penalized(self):
157 self.log.info("Test: outputs above dust threshold are not penalized")
158
159 probe = self.build_tx_with_dust([DUST_THRESHOLD + 100], fee=1000)
160 min_fee = self.get_min_relay_fee(probe)
161
162 tx = self.build_tx_with_dust([DUST_THRESHOLD + 100], fee=min_fee)
163 result = self.nodes[0].testmempoolaccept([tx.serialize().hex()])
164 assert_equal(result[0]["allowed"], True)
165
166
167 if __name__ == '__main__':
168 SubDustFeePenaltyTest(__file__).main()
169