Faster tests with topological mempool rpc sync 🚀 #15858

Could assert error "insufficient fee"?

Could skip when i_remove == i_target?

I opted for code brevity. Calculating the empty set as a difference from two equal sets should take negligible time compared to an rpc call

A brief description of the following algorithm would be handy.

Adding code documentation in the docstring here would be great, detailling use_rpc_sync purpose and use.

It seems slightly odd to me that this code, which we only ever expect to run once, should be in a while loop. I think it'd be clearer to split it out into its own function. Something like:

def sync_mempools(rpc_connections, *, wait=1, timeout=60, use_rpc_sync=False, flush_scheduler=True):
    """
    Wait until everybody has the same transactions in their memory
    pools. If use_rpc_sync is set, sync all transactions right away.
    """
    if use_rpc_sync:
        force_sync_mempools(rpc_connections)
    else:
        stop_time = time.time() + timeout
        while time.time() <= stop_time:
            pool = [set(r.getrawmempool()) for r in rpc_connections]
            if pool.count(pool[0]) == len(rpc_connections):
                break
            time.sleep(wait)
        else:
            raise AssertionError("Mempool sync timed out:{}".format("".join("\n  {!r}".format(m) for m in pool)))

    if flush_scheduler:
        for r in rpc_connections:
            r.syncwithvalidationinterfacequeue()

def force_sync_mempools(rpc_connections):

    class TxInfo:
        def __init__(self, *, raw, ancestors):
            self.raw = raw
            self.ancestors = ancestors

    def topo_send(txs, rpc, pool_add):
        for i in txs:
            topo_send(txs[i].ancestors, rpc, pool_add)
            if i not in pool_add:
                try:
                    assert_equal(i, rpc.sendrawtransaction(txs[i].raw))
                    pool_add.add(i)
                    # Note that conflicted txs (due to RBF) are not removed
                    # from the pool
                except JSONRPCException as e:
                    # This transaction violates policy (e.g. RBF policy). The
                    # mempools should still converge when the high-fee
                    # replacement is synced in a later call
                    assert 'insufficient fee' in e.error['message']

    pool = [set(r.getrawmempool()) for r in rpc_connections]
    # Iterate over all nodes, get their raw mempool and send the
    # missing txs to all other nodes
    for i_remote, rpc_remote in enumerate(rpc_connections):
        pool_remote = {
            txid: TxInfo(raw=rpc_remote.getrawtransaction(txid), ancestors=info['depends'])
            for txid, info in rpc_remote.getrawmempool(verbose=True).items()
        }
        # Create "recursive pools" for ancestors
        for tx in pool_remote:
            pool_remote[tx].ancestors = {a: pool_remote[a] for a in pool_remote[tx].ancestors}

        # Push this pool to all targets
        for i_target, rpc_target in enumerate(rpc_connections):
            missing_txids = pool[i_remote].difference(pool[i_target])
            # Send missing txs
            topo_send(
                txs={txid: pool_remote[txid]
                     for txid in pool_remote if txid in missing_txids},
                rpc=rpc_target,
                pool_add=pool[i_target],
            )
    # If the sync fails there is a logic error in the sync or test code
    pool = [set(r.getrawmempool()) for r in rpc_connections]
    assert pool.count(pool[0]) == len(rpc_connections)

I'd rather not use a while-else, which I have to look up what it does every time I encounter it. I opted for a smaller diff and added an assert, so the diff should be easier to review and it should be clear that it is only run once via the assert.

Also rewriting the function would invalidate previous review

I'm surprised you think Python's while-else syntax is confusing. I don't think I've ever had a problem with it: https://stackoverflow.com/a/3295949. I certainly think it's clearer than the existing code, which has a return buried in a conditional in the while loop, but hidden under the flush_scheduler cleanup code.

I'm pretty sure you've been pushed into this weird construction (execute-once code within a while loop) because of the flush_scheduler cleanup code being inside the loop. Restructuring the function so the loop breaks and then does the cleanup allows the RPC sync to be separated from the while loop, which seems clearer to me.

Unlike if-else, where only one branch is executed, the while-else (or for-else) generally execute both branches. To make it even more confusing, it won't execute the else branch when you break.

Anyway, I am happy to review a follow-up pull that switches both sync_ helpers to the while-else syntax. But, I'd rather not do it here.

nit: remote and target don't seem like very descriptive names to me from and to seem more intuitive to me as the node you're sync'ing from and to.

Renamed to to and from

nit: can you make this comment more explicit? What is a "recursive pool"?

Removed mention of "recursive"

Curious, as the improvement is mostly gained by bypassing Poisson feature, why don't disable it via parameter e.,g,-enablepoisson=0, which would be allowed only in regtest? In this case, nodes behave closer to a real scenario, we have cleaner logs as txs are exchanged via P2P only (no race condition), would work even if sync_mempool hasn't been called and no need to maintain a custom tx exchange function.

@kostyantyn Good point! Done in

• net: Send txs without delay on regtest #15881