interface_usdt_mempool.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2022-present 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
6 """ Tests the mempool:* tracepoint API interface.
7 See https://github.com/limenka/limenka/blob/master/doc/tracing.md#context-mempool
8 """
9
10 import ctypes
11 from decimal import Decimal
12
13 # Test will be skipped if we don't have bcc installed
14 try:
15 from bcc import BPF, USDT # type: ignore[import]
16 except ImportError:
17 pass
18
19 from test_framework.blocktools import COINBASE_MATURITY
20 from test_framework.messages import COIN, DEFAULT_MEMPOOL_EXPIRY_HOURS
21 from test_framework.p2p import P2PDataStore
22 from test_framework.test_framework import LimenkaTestFramework
23 from test_framework.util import (
24 assert_equal,
25 bpf_cflags,
26 )
27 from test_framework.wallet import MiniWallet
28
29 MEMPOOL_TRACEPOINTS_PROGRAM = """
30 # include <uapi/linux/ptrace.h>
31
32 // The longest rejection reason is 118 chars and is generated in case of SCRIPT_ERR_EVAL_FALSE by
33 // strprintf("mandatory-script-verify-flag-failed (%s)", ScriptErrorString(check.GetScriptError()))
34 #define MAX_REJECT_REASON_LENGTH 118
35 // The longest string returned by RemovalReasonToString() is 'sizelimit'
36 #define MAX_REMOVAL_REASON_LENGTH 9
37 #define HASH_LENGTH 32
38
39 struct added_event
40 {
41 u8 hash[HASH_LENGTH];
42 s32 vsize;
43 s64 fee;
44 };
45
46 struct removed_event
47 {
48 u8 hash[HASH_LENGTH];
49 char reason[MAX_REMOVAL_REASON_LENGTH];
50 s32 vsize;
51 s64 fee;
52 u64 entry_time;
53 };
54
55 struct rejected_event
56 {
57 u8 hash[HASH_LENGTH];
58 char reason[MAX_REJECT_REASON_LENGTH];
59 };
60
61 struct replaced_event
62 {
63 u8 replaced_hash[HASH_LENGTH];
64 s32 replaced_vsize;
65 s64 replaced_fee;
66 u64 replaced_entry_time;
67 u8 replacement_hash[HASH_LENGTH];
68 s32 replacement_vsize;
69 s64 replacement_fee;
70 bool replaced_by_transaction;
71 };
72
73 // BPF perf buffer to push the data to user space.
74 BPF_PERF_OUTPUT(added_events);
75 BPF_PERF_OUTPUT(removed_events);
76 BPF_PERF_OUTPUT(rejected_events);
77 BPF_PERF_OUTPUT(replaced_events);
78
79 int trace_added(struct pt_regs *ctx) {
80 struct added_event added = {};
81 void *phash = NULL;
82 bpf_usdt_readarg(1, ctx, &phash);
83 bpf_probe_read_user(&added.hash, sizeof(added.hash), phash);
84 bpf_usdt_readarg(2, ctx, &added.vsize);
85 bpf_usdt_readarg(3, ctx, &added.fee);
86
87 added_events.perf_submit(ctx, &added, sizeof(added));
88 return 0;
89 }
90
91 int trace_removed(struct pt_regs *ctx) {
92 struct removed_event removed = {};
93 void *phash = NULL, *preason = NULL;
94 bpf_usdt_readarg(1, ctx, &phash);
95 bpf_probe_read_user(&removed.hash, sizeof(removed.hash), phash);
96 bpf_usdt_readarg(2, ctx, &preason);
97 bpf_probe_read_user_str(&removed.reason, sizeof(removed.reason), preason);
98 bpf_usdt_readarg(3, ctx, &removed.vsize);
99 bpf_usdt_readarg(4, ctx, &removed.fee);
100 bpf_usdt_readarg(5, ctx, &removed.entry_time);
101
102 removed_events.perf_submit(ctx, &removed, sizeof(removed));
103 return 0;
104 }
105
106 int trace_rejected(struct pt_regs *ctx) {
107 struct rejected_event rejected = {};
108 void *phash = NULL, *preason = NULL;
109 bpf_usdt_readarg(1, ctx, &phash);
110 bpf_probe_read_user(&rejected.hash, sizeof(rejected.hash), phash);
111 bpf_usdt_readarg(2, ctx, &preason);
112 bpf_probe_read_user_str(&rejected.reason, sizeof(rejected.reason), preason);
113 rejected_events.perf_submit(ctx, &rejected, sizeof(rejected));
114 return 0;
115 }
116
117 int trace_replaced(struct pt_regs *ctx) {
118 struct replaced_event replaced = {};
119 void *preplaced_hash = NULL, *preplacement_hash = NULL;
120 bpf_usdt_readarg(1, ctx, &preplaced_hash);
121 bpf_probe_read_user(&replaced.replaced_hash, sizeof(replaced.replaced_hash), preplaced_hash);
122 bpf_usdt_readarg(2, ctx, &replaced.replaced_vsize);
123 bpf_usdt_readarg(3, ctx, &replaced.replaced_fee);
124 bpf_usdt_readarg(4, ctx, &replaced.replaced_entry_time);
125 bpf_usdt_readarg(5, ctx, &preplacement_hash);
126 bpf_probe_read_user(&replaced.replacement_hash, sizeof(replaced.replacement_hash), preplacement_hash);
127 bpf_usdt_readarg(6, ctx, &replaced.replacement_vsize);
128 bpf_usdt_readarg(7, ctx, &replaced.replacement_fee);
129 bpf_usdt_readarg(8, ctx, &replaced.replaced_by_transaction);
130
131 replaced_events.perf_submit(ctx, &replaced, sizeof(replaced));
132 return 0;
133 }
134
135 """
136
137
138 class MempoolReplaced(ctypes.Structure):
139 _fields_ = [
140 ("replaced_hash", ctypes.c_ubyte * 32),
141 ("replaced_vsize", ctypes.c_int32),
142 ("replaced_fee", ctypes.c_int64),
143 ("replaced_entry_time", ctypes.c_uint64),
144 ("replacement_hash", ctypes.c_ubyte * 32),
145 ("replacement_vsize", ctypes.c_int32),
146 ("replacement_fee", ctypes.c_int64),
147 ("replaced_by_transaction", ctypes.c_bool),
148 ]
149
150
151 class MempoolTracepointTest(LimenkaTestFramework):
152 def set_test_params(self):
153 self.num_nodes = 1
154 self.setup_clean_chain = True
155
156 def skip_test_if_missing_module(self):
157 self.skip_if_platform_not_linux()
158 self.skip_if_no_limenkad_tracepoints()
159 self.skip_if_no_python_bcc()
160 self.skip_if_no_bpf_permissions()
161
162 def added_test(self):
163 """Add a transaction to the mempool and make sure the tracepoint returns
164 the expected txid, vsize, and fee."""
165
166 events = []
167
168 self.log.info("Hooking into mempool:added tracepoint...")
169 node = self.nodes[0]
170 ctx = USDT(pid=node.process.pid)
171 ctx.enable_probe(probe="mempool:added", fn_name="trace_added")
172 bpf = BPF(text=MEMPOOL_TRACEPOINTS_PROGRAM, usdt_contexts=[ctx], debug=0, cflags=bpf_cflags())
173
174 def handle_added_event(_, data, __):
175 events.append(bpf["added_events"].event(data))
176
177 bpf["added_events"].open_perf_buffer(handle_added_event)
178
179 self.log.info("Sending transaction...")
180 fee = Decimal(31200)
181 tx = self.wallet.send_self_transfer(from_node=node, fee=fee / COIN)
182
183 self.log.info("Polling buffer...")
184 bpf.perf_buffer_poll(timeout=200)
185
186 self.log.info("Cleaning up mempool...")
187 self.generate(node, 1)
188
189 self.log.info("Ensuring mempool:added event was handled successfully...")
190 assert_equal(1, len(events))
191 event = events[0]
192 assert_equal(bytes(event.hash)[::-1].hex(), tx["txid"])
193 assert_equal(event.vsize, tx["tx"].get_vsize())
194 assert_equal(event.fee, fee)
195
196 bpf.cleanup()
197 self.generate(self.wallet, 1)
198
199 def removed_test(self):
200 """Expire a transaction from the mempool and make sure the tracepoint returns
201 the expected txid, expiry reason, vsize, and fee."""
202
203 events = []
204
205 self.log.info("Hooking into mempool:removed tracepoint...")
206 node = self.nodes[0]
207 ctx = USDT(pid=node.process.pid)
208 ctx.enable_probe(probe="mempool:removed", fn_name="trace_removed")
209 bpf = BPF(text=MEMPOOL_TRACEPOINTS_PROGRAM, usdt_contexts=[ctx], debug=0, cflags=bpf_cflags())
210
211 def handle_removed_event(_, data, __):
212 events.append(bpf["removed_events"].event(data))
213
214 bpf["removed_events"].open_perf_buffer(handle_removed_event)
215
216 self.log.info("Sending transaction...")
217 fee = Decimal(31200)
218 tx = self.wallet.send_self_transfer(from_node=node, fee=fee / COIN)
219 txid = tx["txid"]
220
221 self.log.info("Fast-forwarding time to mempool expiry...")
222 entry_time = node.getmempoolentry(txid)["time"]
223 expiry_time = entry_time + 60 * 60 * DEFAULT_MEMPOOL_EXPIRY_HOURS + 5
224 node.setmocktime(expiry_time)
225
226 self.log.info("Triggering expiry...")
227 self.wallet.get_utxo(txid=txid)
228 self.wallet.send_self_transfer(from_node=node)
229
230 self.log.info("Polling buffer...")
231 bpf.perf_buffer_poll(timeout=200)
232
233 self.log.info("Ensuring mempool:removed event was handled successfully...")
234 assert_equal(1, len(events))
235 event = events[0]
236 assert_equal(bytes(event.hash)[::-1].hex(), txid)
237 assert_equal(event.reason.decode("UTF-8"), "expiry")
238 assert_equal(event.vsize, tx["tx"].get_vsize())
239 assert_equal(event.fee, fee)
240 assert_equal(event.entry_time, entry_time)
241
242 bpf.cleanup()
243 self.generate(self.wallet, 1)
244
245 def replaced_test(self):
246 """Replace one and two transactions in the mempool and make sure the tracepoint
247 returns the expected txids, vsizes, and fees."""
248
249 events = []
250
251 self.log.info("Hooking into mempool:replaced tracepoint...")
252 node = self.nodes[0]
253 ctx = USDT(pid=node.process.pid)
254 ctx.enable_probe(probe="mempool:replaced", fn_name="trace_replaced")
255 bpf = BPF(text=MEMPOOL_TRACEPOINTS_PROGRAM, usdt_contexts=[ctx], debug=0, cflags=bpf_cflags())
256
257 def handle_replaced_event(_, data, __):
258 event = ctypes.cast(data, ctypes.POINTER(MempoolReplaced)).contents
259 events.append(event)
260
261 bpf["replaced_events"].open_perf_buffer(handle_replaced_event)
262
263 self.log.info("Sending RBF transaction...")
264 utxo = self.wallet.get_utxo(mark_as_spent=True)
265 original_fee = Decimal(40000)
266 original_tx = self.wallet.send_self_transfer(
267 from_node=node, utxo_to_spend=utxo, fee=original_fee / COIN
268 )
269 entry_time = node.getmempoolentry(original_tx["txid"])["time"]
270
271 self.log.info("Sending replacement transaction...")
272 replacement_fee = Decimal(45000)
273 replacement_tx = self.wallet.send_self_transfer(
274 from_node=node, utxo_to_spend=utxo, fee=replacement_fee / COIN
275 )
276
277 self.log.info("Polling buffer...")
278 bpf.perf_buffer_poll(timeout=200)
279
280 self.log.info("Ensuring mempool:replaced event was handled successfully...")
281 assert_equal(1, len(events))
282 event = events[0]
283 assert_equal(bytes(event.replaced_hash)[::-1].hex(), original_tx["txid"])
284 assert_equal(event.replaced_vsize, original_tx["tx"].get_vsize())
285 assert_equal(event.replaced_fee, original_fee)
286 assert_equal(event.replaced_entry_time, entry_time)
287 assert_equal(bytes(event.replacement_hash)[::-1].hex(), replacement_tx["txid"])
288 assert_equal(event.replacement_vsize, replacement_tx["tx"].get_vsize())
289 assert_equal(event.replacement_fee, replacement_fee)
290 assert_equal(event.replaced_by_transaction, True)
291
292 bpf.cleanup()
293 self.generate(self.wallet, 1)
294
295 def rejected_test(self):
296 """Create an invalid transaction and make sure the tracepoint returns
297 the expected txid, rejection reason, peer id, and peer address."""
298
299 events = []
300
301 self.log.info("Adding P2P connection...")
302 node = self.nodes[0]
303 node.add_p2p_connection(P2PDataStore())
304
305 self.log.info("Hooking into mempool:rejected tracepoint...")
306 ctx = USDT(pid=node.process.pid)
307 ctx.enable_probe(probe="mempool:rejected", fn_name="trace_rejected")
308 bpf = BPF(text=MEMPOOL_TRACEPOINTS_PROGRAM, usdt_contexts=[ctx], debug=0, cflags=bpf_cflags())
309
310 def handle_rejected_event(_, data, __):
311 events.append(bpf["rejected_events"].event(data))
312
313 bpf["rejected_events"].open_perf_buffer(handle_rejected_event)
314
315 self.log.info("Sending invalid transaction...")
316 tx = self.wallet.create_self_transfer(fee_rate=Decimal(0))
317 node.p2ps[0].send_txs_and_test([tx["tx"]], node, success=False)
318
319 self.log.info("Polling buffer...")
320 bpf.perf_buffer_poll(timeout=200)
321
322 self.log.info("Ensuring mempool:rejected event was handled successfully...")
323 assert_equal(1, len(events))
324 event = events[0]
325 assert_equal(bytes(event.hash)[::-1].hex(), tx["tx"].hash)
326 assert_equal(event.reason.decode("UTF-8"), "min relay fee not met")
327
328 bpf.cleanup()
329 self.generate(self.wallet, 1)
330
331 def run_test(self):
332 """Tests the mempool:added, mempool:removed, mempool:replaced,
333 and mempool:rejected tracepoints."""
334
335 # Create some coinbase transactions and mature them so they can be spent
336 node = self.nodes[0]
337 self.wallet = MiniWallet(node)
338 self.generate(self.wallet, 4)
339 self.generate(node, COINBASE_MATURITY)
340
341 # Test individual tracepoints
342 self.added_test()
343 self.removed_test()
344 self.replaced_test()
345 self.rejected_test()
346
347
348 if __name__ == "__main__":
349 MempoolTracepointTest(__file__).main()
350