Inside Bitcoin

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']