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