interface_zmq.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2015-2022 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 """Test the ZMQ notification interface."""
6
7 import io
8 import os
9 import struct
10 import tempfile
11 from io import BytesIO
12
13 from test_framework.address import (
14 ADDRESS_BCRT1_P2WSH_OP_TRUE,
15 ADDRESS_BCRT1_UNSPENDABLE,
16 )
17 from test_framework.blocktools import (
18 add_witness_commitment,
19 create_block,
20 create_coinbase,
21 )
22 from test_framework.test_framework import LimenkaTestFramework
23 from test_framework.messages import (
24 CBlock,
25 hash256,
26 tx_from_hex,
27 CTransaction,
28 )
29 from test_framework.util import (
30 assert_equal,
31 assert_raises_rpc_error,
32 ensure_for,
33 p2p_port,
34 )
35 from test_framework.wallet import (
36 MiniWallet,
37 )
38 from test_framework.netutil import test_ipv6_local, test_unix_socket
39
40
41 # Test may be skipped and not have zmq installed
42 try:
43 import zmq
44 except ImportError:
45 pass
46
47 def hash256_reversed(byte_str):
48 return hash256(byte_str)[::-1]
49
50 class ZMQSubscriber:
51 def __init__(self, socket, topic):
52 self.sequence = None # no sequence number received yet
53 self.socket = socket
54 self.topic = topic
55
56 self.socket.setsockopt(zmq.SUBSCRIBE, self.topic)
57
58 # Receive message from publisher and verify that topic and sequence match
59 def _receive_from_publisher_and_check(self, specific_topic = None):
60 expected_topic = specific_topic if specific_topic else self.topic
61
62 topic, body, seq = self.socket.recv_multipart()
63 # Topic should match the subscriber topic.
64 assert_equal(topic, expected_topic)
65 # Sequence should be incremental.
66 received_seq = struct.unpack('<I', seq)[-1]
67 if self.sequence is None:
68 self.sequence = received_seq
69 else:
70 assert_equal(received_seq, self.sequence)
71 self.sequence += 1
72 return body
73
74 def receive(self, specific_topic = None):
75 return self._receive_from_publisher_and_check(specific_topic)
76
77 def receive_sequence(self):
78 body = self._receive_from_publisher_and_check()
79 hash = body[:32].hex()
80 label = chr(body[32])
81 mempool_sequence = None if len(body) != 32+1+8 else struct.unpack("<Q", body[32+1:])[0]
82 if mempool_sequence is not None:
83 assert label == "A" or label == "R"
84 else:
85 assert label == "D" or label == "C"
86 return (hash, label, mempool_sequence)
87
88
89 class ZMQTestSetupBlock:
90 """Helper class for setting up a ZMQ test via the "sync up" procedure.
91 Generates a block on the specified node on instantiation and provides a
92 method to check whether a ZMQ notification matches, i.e. the event was
93 caused by this generated block. Assumes that a notification either contains
94 the generated block's hash, it's (coinbase) transaction id, the raw block or
95 raw transaction data.
96 """
97 def __init__(self, test_framework, node):
98 if test_framework.is_wallet_compiled():
99 self.block_hash = test_framework.generatetoaddress(node, nblocks=1, address=node.getnewaddress(), maxtries=1000000, sync_fun=test_framework.no_op)[0]
100 else:
101 self.block_hash = test_framework.generate(node, 1, sync_fun=test_framework.no_op)[0]
102 coinbase = node.getblock(self.block_hash, 2)['tx'][0]
103 self.tx_hash = coinbase['txid']
104 self.raw_tx = coinbase['hex']
105 self.raw_block = node.getblock(self.block_hash, 0)
106
107 def caused_notification(self, notification):
108 return (
109 self.block_hash in notification
110 or self.tx_hash in notification
111 or self.raw_block in notification
112 or self.raw_tx in notification
113 )
114
115
116 class ZMQTest (LimenkaTestFramework):
117 def add_options(self, parser):
118 self.add_wallet_options(parser)
119
120 def set_test_params(self):
121 self.num_nodes = 2
122 # whitelist peers to speed up tx relay / mempool sync
123 self.noban_tx_relay = True
124 self.zmq_port_base = p2p_port(self.num_nodes + 1)
125 if self.is_wallet_compiled():
126 self.skip_if_no_wallet()
127
128 def skip_test_if_missing_module(self):
129 self.skip_if_no_py3_zmq()
130 self.skip_if_no_limenkad_zmq()
131
132 def run_test(self):
133 self.wallet = MiniWallet(self.nodes[0])
134 self.ctx = zmq.Context()
135 try:
136 self.test_basic()
137 if test_unix_socket():
138 self.test_basic(unix=True)
139 else:
140 self.log.info("Skipping ipc test, because UNIX sockets are not supported.")
141 self.test_sequence()
142 self.test_mempool_sync()
143 self.test_reorg()
144 self.test_multiple_interfaces()
145 self.test_ipv6()
146 finally:
147 # Destroy the ZMQ context.
148 self.log.debug("Destroying ZMQ context")
149 self.ctx.destroy(linger=None)
150
151 # Restart node with the specified zmq notifications enabled, subscribe to
152 # all of them and return the corresponding ZMQSubscriber objects.
153 def setup_zmq_test(self, services, *, recv_timeout=60, sync_blocks=True, ipv6=False):
154 subscribers = []
155 for topic, address in services:
156 socket = self.ctx.socket(zmq.SUB)
157 if ipv6:
158 socket.setsockopt(zmq.IPV6, 1)
159 subscribers.append(ZMQSubscriber(socket, topic.encode()))
160
161 self.restart_node(0, [f"-zmqpub{topic}={address.replace('ipc://', 'unix:')}" for topic, address in services])
162
163 for i, sub in enumerate(subscribers):
164 sub.socket.connect(services[i][1])
165
166 # Ensure that all zmq publisher notification interfaces are ready by
167 # running the following "sync up" procedure:
168 # 1. Generate a block on the node
169 # 2. Try to receive the corresponding notification on all subscribers
170 # 3. If all subscribers get the message within the timeout (1 second),
171 # we are done, otherwise repeat starting from step 1
172 for sub in subscribers:
173 sub.socket.set(zmq.RCVTIMEO, 1000)
174 while True:
175 test_block = ZMQTestSetupBlock(self, self.nodes[0])
176 recv_failed = False
177 for sub in subscribers:
178 try:
179 specific_topic = sub.topic
180 if b'wallet' in sub.topic:
181 specific_topic += b"-block"
182 while not test_block.caused_notification(sub.receive(specific_topic=specific_topic).hex()):
183 self.log.debug("Ignoring sync-up notification for previously generated block.")
184 except zmq.error.Again:
185 self.log.debug("Didn't receive sync-up notification, trying again.")
186 recv_failed = True
187 if not recv_failed:
188 self.log.debug("ZMQ sync-up completed, all subscribers are ready.")
189 break
190
191 # set subscriber's desired timeout for the test
192 for sub in subscribers:
193 sub.socket.set(zmq.RCVTIMEO, recv_timeout*1000)
194
195 self.connect_nodes(0, 1)
196 if sync_blocks:
197 self.sync_blocks()
198
199 return subscribers
200
201 def test_basic(self, unix = False):
202 self.log.info(f"Running basic test with {'ipc' if unix else 'tcp'} protocol")
203
204 # Invalid zmq arguments don't take down the node, see #17185.
205 self.restart_node(0, ["-zmqpubrawtx=foo", "-zmqpubhashtx=bar"])
206
207 address = f"tcp://127.0.0.1:{self.zmq_port_base}"
208
209 if unix:
210 # Use the shortest temp path possible since paths may have as little as 92-char limit
211 socket_path = tempfile.NamedTemporaryFile().name
212 address = f"ipc://{socket_path}"
213
214 services = ["hashblock", "hashtx", "rawblock", "rawtx"]
215 if self.is_wallet_compiled():
216 services += ["hashwallettx", "rawwallettx"]
217 subs = self.setup_zmq_test([(topic, address) for topic in services])
218
219 hashblock = subs[0]
220 hashtx = subs[1]
221 rawblock = subs[2]
222 rawtx = subs[3]
223 if self.is_wallet_compiled():
224 hashwallettx = subs[-2]
225 rawwallettx = subs[-1]
226
227 self.sync_all()
228 # Flush initial wallettx events before we begin
229 while True:
230 try:
231 topic, body, seq = hashwallettx.socket.recv_multipart()
232 except zmq.ZMQError:
233 break
234 subscriber = {b'hashwallettx-block': hashwallettx, b'rawwallettx-block': rawwallettx}[topic]
235 assert_equal(struct.unpack('<I', seq)[-1], subscriber.sequence)
236 subscriber.sequence += 1
237
238 num_blocks = 5
239 self.log.info(f"Generate {num_blocks} blocks (and {num_blocks} coinbase txes)")
240 if self.is_wallet_compiled():
241 genhashes = self.generate(self.nodes[0], num_blocks)
242 else:
243 genhashes = self.generatetoaddress(self.nodes[0], num_blocks, ADDRESS_BCRT1_UNSPENDABLE)
244
245 for x in range(num_blocks):
246 # Should receive the coinbase txid.
247 txid = hashtx.receive()
248
249 # Should receive the coinbase raw transaction.
250 tx = tx_from_hex(rawtx.receive().hex())
251 tx.calc_sha256()
252 assert_equal(tx.hash, txid.hex())
253
254 # Should receive the generated raw block.
255 hex = rawblock.receive()
256 block = CBlock()
257 block.deserialize(BytesIO(hex))
258 assert block.is_valid()
259 assert_equal(block.vtx[0].hash, tx.hash)
260 assert_equal(len(block.vtx), 1)
261 assert_equal(genhashes[x], hash256_reversed(hex[:80]).hex())
262
263 if self.is_wallet_compiled():
264 # Should receive wallet tx
265 wallettxid = hashwallettx.receive(b"hashwallettx-block")
266 wallethex = rawwallettx.receive(b"rawwallettx-block")
267 wallettx = CTransaction()
268 wallettx.deserialize(io.BytesIO(wallethex))
269 wallettx.calc_sha256()
270 assert_equal(wallettx.hash, wallettxid.hex())
271
272 # Should receive the generated block hash.
273 hash = hashblock.receive().hex()
274 assert_equal(genhashes[x], hash)
275 # The block should only have the coinbase txid.
276 assert_equal([txid.hex()], self.nodes[1].getblock(hash)["tx"])
277
278
279 self.log.info("Wait for tx from second node")
280 if self.is_wallet_compiled():
281 payment_txid = self.nodes[1].sendtoaddress(self.nodes[0].getnewaddress(), 1.0)
282 payment_tx = {'wtxid': self.nodes[1].getrawtransaction(payment_txid, 1)['hash']}
283 else:
284 payment_tx = self.wallet.send_self_transfer(from_node=self.nodes[1])
285 payment_txid = payment_tx['txid']
286 self.sync_all()
287 # Should receive the broadcasted txid.
288 txid = hashtx.receive()
289 assert_equal(payment_txid, txid.hex())
290
291 # Should receive the broadcasted raw transaction.
292 hex = rawtx.receive()
293 assert_equal(payment_tx['wtxid'], hash256_reversed(hex).hex())
294
295 # Mining the block with this tx should result in second notification
296 # after coinbase tx notification
297 self.generatetoaddress(self.nodes[0], 1, ADDRESS_BCRT1_UNSPENDABLE)
298 hashtx.receive()
299 txid = hashtx.receive()
300 assert_equal(payment_txid, txid.hex())
301
302 if self.is_wallet_compiled():
303 wallettxid = hashwallettx.receive(b"hashwallettx-mempool")
304 wallethex = rawwallettx.receive(b"rawwallettx-mempool")
305 wallettx = CTransaction()
306 wallettx.deserialize(io.BytesIO(wallethex))
307 wallettx.calc_sha256()
308 assert_equal(wallettx.hash, wallettxid.hex())
309
310 self.log.info("Test the getzmqnotifications RPC")
311 assert_equal(self.nodes[0].getzmqnotifications(), [
312 {"type": "pubhashblock", "address": address, "hwm": 1000},
313 {"type": "pubhashtx", "address": address, "hwm": 1000},
314 ] + ([{"type": "pubhashwallettx", "address": address, "hwm": 1000}] if self.is_wallet_compiled() else []) + [
315 {"type": "pubrawblock", "address": address, "hwm": 1000},
316 {"type": "pubrawtx", "address": address, "hwm": 1000},
317 ] + ([{"type": "pubrawwallettx", "address": address, "hwm": 1000}] if self.is_wallet_compiled() else []) + [
318 ])
319
320 assert_equal(self.nodes[1].getzmqnotifications(), [])
321 if unix:
322 os.unlink(socket_path)
323
324 def test_reorg(self):
325
326 address = f"tcp://127.0.0.1:{self.zmq_port_base}"
327
328 # Should only notify the tip if a reorg occurs
329 hashblock, hashtx = self.setup_zmq_test(
330 [(topic, address) for topic in ["hashblock", "hashtx"]],
331 recv_timeout=2) # 2 second timeout to check end of notifications
332 self.disconnect_nodes(0, 1)
333
334 # Generate 1 block in nodes[0] with 1 mempool tx and receive all notifications
335 payment_txid = self.wallet.send_self_transfer(from_node=self.nodes[0])['txid']
336 disconnect_block = self.generatetoaddress(self.nodes[0], 1, ADDRESS_BCRT1_UNSPENDABLE, sync_fun=self.no_op)[0]
337 disconnect_cb = self.nodes[0].getblock(disconnect_block)["tx"][0]
338 assert_equal(self.nodes[0].getbestblockhash(), hashblock.receive().hex())
339 assert_equal(hashtx.receive().hex(), payment_txid)
340 assert_equal(hashtx.receive().hex(), disconnect_cb)
341
342 # Generate 2 blocks in nodes[1] to a different address to ensure split
343 connect_blocks = self.generatetoaddress(self.nodes[1], 2, ADDRESS_BCRT1_P2WSH_OP_TRUE, sync_fun=self.no_op)
344
345 # nodes[0] will reorg chain after connecting back nodes[1]
346 self.connect_nodes(0, 1)
347 self.sync_blocks() # tx in mempool valid but not advertised
348
349 # Should receive nodes[1] tip
350 assert_equal(self.nodes[1].getbestblockhash(), hashblock.receive().hex())
351
352 # During reorg:
353 # Get old payment transaction notification from disconnect and disconnected cb
354 assert_equal(hashtx.receive().hex(), payment_txid)
355 assert_equal(hashtx.receive().hex(), disconnect_cb)
356 # And the payment transaction again due to mempool entry
357 assert_equal(hashtx.receive().hex(), payment_txid)
358 assert_equal(hashtx.receive().hex(), payment_txid)
359 # And the new connected coinbases
360 for i in [0, 1]:
361 assert_equal(hashtx.receive().hex(), self.nodes[1].getblock(connect_blocks[i])["tx"][0])
362
363 # If we do a simple invalidate we announce the disconnected coinbase
364 self.nodes[0].invalidateblock(connect_blocks[1])
365 assert_equal(hashtx.receive().hex(), self.nodes[1].getblock(connect_blocks[1])["tx"][0])
366 # And the current tip
367 assert_equal(hashtx.receive().hex(), self.nodes[1].getblock(connect_blocks[0])["tx"][0])
368
369 def test_sequence(self):
370 """
371 Sequence zmq notifications give every blockhash and txhash in order
372 of processing, regardless of IBD, re-orgs, etc.
373 Format of messages:
374 <32-byte hash>C : Blockhash connected
375 <32-byte hash>D : Blockhash disconnected
376 <32-byte hash>R<8-byte LE uint> : Transactionhash removed from mempool for non-block inclusion reason
377 <32-byte hash>A<8-byte LE uint> : Transactionhash added mempool
378 """
379 self.log.info("Testing 'sequence' publisher")
380 [seq] = self.setup_zmq_test([("sequence", f"tcp://127.0.0.1:{self.zmq_port_base}")])
381 self.disconnect_nodes(0, 1)
382
383 # Mempool sequence number starts at 1
384 seq_num = 1
385
386 # Generate 1 block in nodes[0] and receive all notifications
387 dc_block = self.generatetoaddress(self.nodes[0], 1, ADDRESS_BCRT1_UNSPENDABLE, sync_fun=self.no_op)[0]
388
389 # Note: We are not notified of any block transactions, coinbase or mined
390 assert_equal((self.nodes[0].getbestblockhash(), "C", None), seq.receive_sequence())
391
392 # Generate 2 blocks in nodes[1] to a different address to ensure a chain split
393 self.generatetoaddress(self.nodes[1], 2, ADDRESS_BCRT1_P2WSH_OP_TRUE, sync_fun=self.no_op)
394
395 # nodes[0] will reorg chain after connecting back nodes[1]
396 self.connect_nodes(0, 1)
397
398 # Then we receive all block (dis)connect notifications for the 2 block reorg
399 assert_equal((dc_block, "D", None), seq.receive_sequence())
400 block_count = self.nodes[1].getblockcount()
401 assert_equal((self.nodes[1].getblockhash(block_count-1), "C", None), seq.receive_sequence())
402 assert_equal((self.nodes[1].getblockhash(block_count), "C", None), seq.receive_sequence())
403
404 self.log.info("Wait for tx from second node")
405 payment_tx = self.wallet.send_self_transfer(from_node=self.nodes[1])
406 payment_txid = payment_tx['txid']
407 self.sync_all()
408 self.log.info("Testing sequence notifications with mempool sequence values")
409
410 # Should receive the broadcasted txid.
411 assert_equal((payment_txid, "A", seq_num), seq.receive_sequence())
412 seq_num += 1
413
414 self.log.info("Testing RBF notification")
415 # Replace it to test eviction/addition notification
416 payment_tx['tx'].vout[0].nValue -= 1000
417 rbf_txid = self.nodes[1].sendrawtransaction(payment_tx['tx'].serialize().hex())
418 self.sync_all()
419 assert_equal((payment_txid, "R", seq_num), seq.receive_sequence())
420 seq_num += 1
421 assert_equal((rbf_txid, "A", seq_num), seq.receive_sequence())
422 seq_num += 1
423
424 # Doesn't get published when mined, make a block and tx to "flush" the possibility
425 # though the mempool sequence number does go up by the number of transactions
426 # removed from the mempool by the block mining it.
427 mempool_size = len(self.nodes[0].getrawmempool())
428 c_block = self.generatetoaddress(self.nodes[0], 1, ADDRESS_BCRT1_UNSPENDABLE)[0]
429 # Make sure the number of mined transactions matches the number of txs out of mempool
430 mempool_size_delta = mempool_size - len(self.nodes[0].getrawmempool())
431 assert_equal(len(self.nodes[0].getblock(c_block)["tx"])-1, mempool_size_delta)
432 seq_num += mempool_size_delta
433 payment_txid_2 = self.wallet.send_self_transfer(from_node=self.nodes[1])['txid']
434 self.sync_all()
435 assert_equal((c_block, "C", None), seq.receive_sequence())
436 assert_equal((payment_txid_2, "A", seq_num), seq.receive_sequence())
437 seq_num += 1
438
439 # Spot check getrawmempool results that they only show up when asked for
440 assert type(self.nodes[0].getrawmempool()) is list
441 assert type(self.nodes[0].getrawmempool(mempool_sequence=False)) is list
442 assert "mempool_sequence" not in self.nodes[0].getrawmempool(verbose=True)
443 assert_raises_rpc_error(-8, "Verbose results cannot contain mempool sequence values.", self.nodes[0].getrawmempool, True, True)
444 assert_equal(self.nodes[0].getrawmempool(mempool_sequence=True)["mempool_sequence"], seq_num)
445
446 self.log.info("Testing reorg notifications")
447 # Manually invalidate the last block to test mempool re-entry
448 # N.B. This part could be made more lenient in exact ordering
449 # since it greatly depends on inner-workings of blocks/mempool
450 # during "deep" re-orgs. Probably should "re-construct"
451 # blockchain/mempool state from notifications instead.
452 block_count = self.nodes[0].getblockcount()
453 best_hash = self.nodes[0].getbestblockhash()
454 self.nodes[0].invalidateblock(best_hash)
455
456 # Make sure getrawmempool mempool_sequence results aren't "queued" but immediately reflective
457 # of the time they were gathered.
458 ensure_for(duration=2, f=lambda: self.nodes[0].getrawmempool(mempool_sequence=True)["mempool_sequence"] > seq_num)
459
460 assert_equal((best_hash, "D", None), seq.receive_sequence())
461 assert_equal((rbf_txid, "A", seq_num), seq.receive_sequence())
462 seq_num += 1
463
464 # Other things may happen but aren't wallet-deterministic so we don't test for them currently
465 self.nodes[0].reconsiderblock(best_hash)
466 self.generatetoaddress(self.nodes[1], 1, ADDRESS_BCRT1_UNSPENDABLE)
467
468 self.log.info("Evict mempool transaction by block conflict")
469 orig_tx = self.wallet.send_self_transfer(from_node=self.nodes[0])
470 orig_txid = orig_tx['txid']
471
472 # More to be simply mined
473 more_tx = []
474 for _ in range(5):
475 more_tx.append(self.wallet.send_self_transfer(from_node=self.nodes[0]))
476
477 orig_tx['tx'].vout[0].nValue -= 1000
478 bump_txid = self.nodes[0].sendrawtransaction(orig_tx['tx'].serialize().hex())
479 # Mine the pre-bump tx
480 txs_to_add = [orig_tx['hex']] + [tx['hex'] for tx in more_tx]
481 block = create_block(int(self.nodes[0].getbestblockhash(), 16), create_coinbase(self.nodes[0].getblockcount()+1), txlist=txs_to_add)
482 add_witness_commitment(block)
483 block.solve()
484 assert_equal(self.nodes[0].submitblock(block.serialize().hex()), None)
485 tip = self.nodes[0].getbestblockhash()
486 assert_equal(int(tip, 16), block.sha256)
487 orig_txid_2 = self.wallet.send_self_transfer(from_node=self.nodes[0])['txid']
488
489 # Flush old notifications until evicted tx original entry
490 (hash_str, label, mempool_seq) = seq.receive_sequence()
491 while hash_str != orig_txid:
492 (hash_str, label, mempool_seq) = seq.receive_sequence()
493 mempool_seq += 1
494
495 # Added original tx
496 assert_equal(label, "A")
497 # More transactions to be simply mined
498 for i in range(len(more_tx)):
499 assert_equal((more_tx[i]['txid'], "A", mempool_seq), seq.receive_sequence())
500 mempool_seq += 1
501 # Bumped by rbf
502 assert_equal((orig_txid, "R", mempool_seq), seq.receive_sequence())
503 mempool_seq += 1
504 assert_equal((bump_txid, "A", mempool_seq), seq.receive_sequence())
505 mempool_seq += 1
506 # Conflict announced first, then block
507 assert_equal((bump_txid, "R", mempool_seq), seq.receive_sequence())
508 mempool_seq += 1
509 assert_equal((tip, "C", None), seq.receive_sequence())
510 mempool_seq += len(more_tx)
511 # Last tx
512 assert_equal((orig_txid_2, "A", mempool_seq), seq.receive_sequence())
513 mempool_seq += 1
514 self.generatetoaddress(self.nodes[0], 1, ADDRESS_BCRT1_UNSPENDABLE)
515 self.sync_all() # want to make sure we didn't break "consensus" for other tests
516
517 def test_mempool_sync(self):
518 """
519 Use sequence notification plus getrawmempool sequence results to "sync mempool"
520 """
521
522 self.log.info("Testing 'mempool sync' usage of sequence notifier")
523 [seq] = self.setup_zmq_test([("sequence", f"tcp://127.0.0.1:{self.zmq_port_base}")])
524
525 # In-memory counter, should always start at 1
526 next_mempool_seq = self.nodes[0].getrawmempool(mempool_sequence=True)["mempool_sequence"]
527 assert_equal(next_mempool_seq, 1)
528
529 # Some transactions have been happening but we aren't consuming zmq notifications yet
530 # or we lost a ZMQ message somehow and want to start over
531 txs = []
532 num_txs = 5
533 for _ in range(num_txs):
534 txs.append(self.wallet.send_self_transfer(from_node=self.nodes[1]))
535 self.sync_all()
536
537 # 1) Consume backlog until we get a mempool sequence number
538 (hash_str, label, zmq_mem_seq) = seq.receive_sequence()
539 while zmq_mem_seq is None:
540 (hash_str, label, zmq_mem_seq) = seq.receive_sequence()
541
542 assert label == "A" or label == "R"
543 assert hash_str is not None
544
545 # 2) We need to "seed" our view of the mempool
546 mempool_snapshot = self.nodes[0].getrawmempool(mempool_sequence=True)
547 mempool_view = set(mempool_snapshot["txids"])
548 get_raw_seq = mempool_snapshot["mempool_sequence"]
549 assert_equal(get_raw_seq, num_txs + 1)
550 assert zmq_mem_seq < get_raw_seq
551
552 # Things continue to happen in the "interim" while waiting for snapshot results
553 # We have node 0 do all these to avoid p2p races with RBF announcements
554 for _ in range(num_txs):
555 txs.append(self.wallet.send_self_transfer(from_node=self.nodes[0]))
556 txs[-1]['tx'].vout[0].nValue -= 1000
557 self.nodes[0].sendrawtransaction(txs[-1]['tx'].serialize().hex())
558 self.sync_all()
559 self.generatetoaddress(self.nodes[0], 1, ADDRESS_BCRT1_UNSPENDABLE)
560 final_txid = self.wallet.send_self_transfer(from_node=self.nodes[0])['txid']
561
562 # 3) Consume ZMQ backlog until we get to "now" for the mempool snapshot
563 while True:
564 if zmq_mem_seq == get_raw_seq - 1:
565 break
566 (hash_str, label, mempool_sequence) = seq.receive_sequence()
567 if mempool_sequence is not None:
568 zmq_mem_seq = mempool_sequence
569 if zmq_mem_seq > get_raw_seq:
570 raise Exception(f"We somehow jumped mempool sequence numbers! zmq_mem_seq: {zmq_mem_seq} > get_raw_seq: {get_raw_seq}")
571
572 # 4) Moving forward, we apply the delta to our local view
573 # remaining txs(5) + 1 rbf(A+R) + 1 block connect + 1 final tx
574 expected_sequence = get_raw_seq
575 r_gap = 0
576 for _ in range(num_txs + 2 + 1 + 1):
577 (hash_str, label, mempool_sequence) = seq.receive_sequence()
578 if mempool_sequence is not None:
579 if mempool_sequence != expected_sequence:
580 # Detected "R" gap, means this a conflict eviction, and mempool tx are being evicted before its
581 # position in the incoming block message "C"
582 if label == "R":
583 assert mempool_sequence > expected_sequence
584 r_gap += mempool_sequence - expected_sequence
585 else:
586 raise Exception(f"WARNING: txhash has unexpected mempool sequence value: {mempool_sequence} vs expected {expected_sequence}")
587 if label == "A":
588 assert hash_str not in mempool_view
589 mempool_view.add(hash_str)
590 expected_sequence = mempool_sequence + 1
591 elif label == "R":
592 assert hash_str in mempool_view
593 mempool_view.remove(hash_str)
594 expected_sequence = mempool_sequence + 1
595 elif label == "C":
596 # (Attempt to) remove all txids from known block connects
597 block_txids = self.nodes[0].getblock(hash_str)["tx"][1:]
598 for txid in block_txids:
599 if txid in mempool_view:
600 expected_sequence += 1
601 mempool_view.remove(txid)
602 expected_sequence -= r_gap
603 r_gap = 0
604 elif label == "D":
605 # Not useful for mempool tracking per se
606 continue
607 else:
608 raise Exception("Unexpected ZMQ sequence label!")
609
610 assert_equal(self.nodes[0].getrawmempool(), [final_txid])
611 assert_equal(self.nodes[0].getrawmempool(mempool_sequence=True)["mempool_sequence"], expected_sequence)
612
613 # 5) If you miss a zmq/mempool sequence number, go back to step (2)
614
615 self.generatetoaddress(self.nodes[0], 1, ADDRESS_BCRT1_UNSPENDABLE)
616
617 def test_multiple_interfaces(self):
618 # Set up two subscribers with different addresses
619 # (note that after the reorg test, syncing would fail due to different
620 # chain lengths on node0 and node1; for this test we only need node0, so
621 # we can disable syncing blocks on the setup)
622 subscribers = self.setup_zmq_test([
623 ("hashblock", f"tcp://127.0.0.1:{self.zmq_port_base + 1}"),
624 ("hashblock", f"tcp://127.0.0.1:{self.zmq_port_base + 2}"),
625 ], sync_blocks=False)
626
627 # Generate 1 block in nodes[0] and receive all notifications
628 self.generatetoaddress(self.nodes[0], 1, ADDRESS_BCRT1_UNSPENDABLE, sync_fun=self.no_op)
629
630 # Should receive the same block hash on both subscribers
631 assert_equal(self.nodes[0].getbestblockhash(), subscribers[0].receive().hex())
632 assert_equal(self.nodes[0].getbestblockhash(), subscribers[1].receive().hex())
633
634 def test_ipv6(self):
635 if not test_ipv6_local():
636 self.log.info("Skipping IPv6 test, because IPv6 is not supported.")
637 return
638 self.log.info("Testing IPv6")
639 # Set up subscriber using IPv6 loopback address
640 subscribers = self.setup_zmq_test([
641 ("hashblock", f"tcp://[::1]:{self.zmq_port_base}")
642 ], ipv6=True)
643
644 # Generate 1 block in nodes[0]
645 self.generatetoaddress(self.nodes[0], 1, ADDRESS_BCRT1_UNSPENDABLE)
646
647 # Should receive the same block hash
648 assert_equal(self.nodes[0].getbestblockhash(), subscribers[0].receive().hex())
649
650
651 if __name__ == '__main__':
652 ZMQTest(__file__).main()
653