add a test for racy forwarding requests

Changelog-None.

Signed-off-by: Lagrang3 <[email protected]>

tests/test_pay.py

@@ -5996,3 +5996,152 @@ def test_enableoffer(node_factory):
     # Can't enable unknown.
     with pytest.raises(RpcError, match="Unknown offer"):
         l1.rpc.enableoffer(offer_id=offer1['offer_id'])
+
+
+def get_local_channel_by_id(node, chanid):
+    peerchannels = node.rpc.listpeerchannels()["channels"]
+    if not peerchannels:
+        return None
+    for c in peerchannels:
+        if c["channel_id"] == chanid:
+            return c
+    return None
+
+
+def start_channels(connections):
+    """Similar to join_nodes but with arbitrary connections."""
+    nodes = list()
+    for src, dst, fundamount in connections:
+        nodes.append(src)
+        nodes.append(dst)
+        src.rpc.connect(dst.info["id"], "localhost", dst.port)
+
+    bitcoind = nodes[0].bitcoin
+    # If we got here, we want to fund channels
+    for src, dst, fundamount in connections:
+        addr = src.rpc.newaddr()["bech32"]
+        bitcoind.rpc.sendtoaddress(addr, (fundamount + 1000000) / 10**8)
+
+    bitcoind.generate_block(1)
+    sync_blockheight(bitcoind, nodes)
+    txids = []
+    chan_ids = []
+    for src, dst, fundamount in connections:
+        reply = src.rpc.fundchannel(
+            dst.info["id"], fundamount, announce=True, minconf=0
+        )
+        txids.append(reply["txid"])
+        chan_ids.append(reply["channel_id"])
+
+    # Confirm all channels and wait for them to become usable
+    bitcoind.generate_block(1, wait_for_mempool=txids)
+    scids = []
+    for con, mychan_id in zip(connections, chan_ids):
+        src = con[0]
+        wait_for(
+            lambda: get_local_channel_by_id(src, mychan_id)["state"]
+            == "CHANNELD_NORMAL"
+        )
+        scids.append(get_local_channel_by_id(src, mychan_id)["short_channel_id"])
+
+    # Make sure they have all seen block so they don't complain about
+    # the coming gossip messages
+    sync_blockheight(bitcoind, nodes)
+    bitcoind.generate_block(5)
+
+    # Make sure everyone sees all channels, all other nodes
+    for n in nodes:
+        for scid in scids:
+            n.wait_channel_active(scid)
+
+    # Make sure we have all node announcements, too
+    for n in nodes:
+        for n2 in nodes:
+            wait_for(
+                lambda: "alias" in only_one(n.rpc.listnodes(n2.info["id"])["nodes"])
+            )
+    return chan_ids
+
+
+def test_parallel_channels_reserve(node_factory):
+    """Tests wether we are able to pay through parallel channels concurrently."""
+    def direction(node1, node2):
+        return 0 if node1.info["id"] < node2.info["id"] else 1
+
+    opts = {"disable-mpp": None, "fee-base": 0, "fee-per-satoshi": 0, "cltv-delta": 6}
+    l1, l2, l3 = node_factory.get_nodes(3, opts=opts)
+
+    chan_ids = start_channels(
+        [
+            (l1, l2, 100_000),
+            (l2, l3, 100_000),
+            (l1, l2, 200_000),
+            (l2, l3, 200_000),
+            (l1, l2, 300_000),
+            (l2, l3, 300_000),
+            (l1, l2, 400_000),
+            (l2, l3, 400_000),
+        ]
+    )
+
+    # we should be able to send these four parts:
+    nparts = 4
+    route_amounts = ["75000sat", "175000sat", "275000sat", "375000sat"]
+    total_msat = sum([Millisatoshi(a) for a in route_amounts[:nparts]])
+
+    # Test succeeds if we are able to pay this invoice
+    inv = l3.rpc.call(
+        "invoice",
+        {"amount_msat": total_msat, "label": "inv", "description": "inv", "cltv": 10},
+    )
+    inv_decode = l1.rpc.decode(inv["bolt11"])
+
+    # Share data by every route we will construct: l1->l2->l3
+    route = [
+        {
+            "id": l2.info["id"],
+            "direction": direction(l1, l2),
+            "delay": 16,
+            "style": "tlv",
+        },
+        {
+            "id": l3.info["id"],
+            "direction": direction(l2, l3),
+            "delay": 10,
+            "style": "tlv",
+        },
+    ]
+
+    # Send every part with sendpay
+    for part in range(nparts):
+        this_part_msat = Millisatoshi(route_amounts[part])
+        chan1 = get_local_channel_by_id(l1, chan_ids[part * 2])
+        chan2 = get_local_channel_by_id(l2, chan_ids[part * 2 + 1])
+
+        route[0]["channel"] = chan1["short_channel_id"]
+        route[1]["channel"] = chan2["short_channel_id"]
+        route[0]["amount_msat"] = route[1]["amount_msat"] = this_part_msat
+
+        assert chan1["spendable_msat"] >= this_part_msat
+        assert chan2["spendable_msat"] >= this_part_msat
+
+        l1.rpc.call(
+            "sendpay",
+            {
+                "route": route,
+                "payment_hash": inv["payment_hash"],
+                "payment_secret": inv["payment_secret"],
+                "amount_msat": total_msat,
+                "groupid": 1,
+                "partid": part + 1,
+            },
+        )
+    l1.wait_for_htlcs()
+
+    # Are we happy?
+    waitsendpay_response = l1.rpc.call(
+        "waitsendpay", {"payment_hash": inv["payment_hash"], "partid": 1, "groupid": 1}
+    )
+    receipt = only_one(l3.rpc.listinvoices("inv")["invoices"])
+    assert receipt["status"] == "paid"
+    assert receipt["amount_received_msat"] == total_msat