scenarios: improve ln_init for large graphs with channel polciies

Author: pinheadmz

Diff for: src/scenarios/ln_init.py

@@ -1,8 +1,10 @@
 #!/usr/bin/env python3
+
 from time import sleep
 
 from scenarios.utils import ensure_miner
 from warnet.test_framework_bridge import WarnetTestFramework
+from warnet.utils import channel_match
 
 
 def cli_help():
@@ -15,77 +17,127 @@ def set_test_params(self):
 
     def run_test(self):
         self.log.info("Get LN nodes and wallet addresses")
+        ln_nodes = []
         recv_addrs = []
         for tank in self.warnet.tanks:
             if tank.lnnode is not None:
                 recv_addrs.append(tank.lnnode.getnewaddress())
+                ln_nodes.append(tank.index)
 
         self.log.info("Fund LN wallets")
-        miner = ensure_miner(self.nodes[3])
-        addr = miner.getnewaddress()
-        self.generatetoaddress(self.nodes[3], 110, addr)
+        miner = ensure_miner(self.nodes[0])
+        miner_addr = miner.getnewaddress()
+        # 298 block base
+        self.generatetoaddress(self.nodes[0], 298, miner_addr)
+        # divvy up the goods
+        split = miner.getbalance() // len(recv_addrs)
         for addr in recv_addrs:
-            miner.sendtoaddress(addr, 50)
-        self.generatetoaddress(self.nodes[3], 1, addr)
-
-        self.log.info("Waiting for funds to be spendable")
-        ready = False
-        while not ready:
-            sleep(1)
-            ready = True
+            miner.sendtoaddress(addr, split)
+        # confirm funds in block 299
+        self.generatetoaddress(self.nodes[0], 1, miner_addr)
+
+        self.log.info(f"Waiting for funds to be spendable: {split} BTC each for {len(recv_addrs)} LN nodes")
+
+        def funded_lnnodes():
             for tank in self.warnet.tanks:
                 if tank.lnnode is None:
                     continue
-                bal = tank.lnnode.get_wallet_balance()
-                if int(bal["confirmed_balance"]) >= 50:
-                    continue
-                ready = False
-                break
+                if int(tank.lnnode.get_wallet_balance()["confirmed_balance"]) < (split * 100000000):
+                    return False
+            return True
+        self.wait_until(funded_lnnodes, timeout=5*60)
+
+        ln_nodes_uri = ln_nodes.copy()
+        while len(ln_nodes_uri) > 0:
+            self.log.info(f"Waiting for all LN nodes to have URI, LN nodes remaining: {ln_nodes_uri}")
+            for index in ln_nodes_uri:
+                lnnode = self.warnet.tanks[index].lnnode
+                if lnnode.getURI():
+                    ln_nodes_uri.remove(index)
+            sleep(5)
 
-        self.log.info("Open channels")
-        # TODO: This might belong in Warnet class as connect_ln_edges()
-        #       but that would need to ensure spendable funds first.
-        #       For now we consider this scenario "special".
-        opening_txs = []
-        ln_edges = []
+        self.log.info("Adding p2p connections to LN nodes")
         for edge in self.warnet.graph.edges(data=True):
             (src, dst, data) = edge
-            if "channel" in data:
+            # Copy the L1 p2p topology (where applicable) to L2
+            # so we get a more robust p2p graph for lightning
+            if "channel_open" not in data and self.warnet.tanks[src].lnnode and self.warnet.tanks[dst].lnnode:
+                self.warnet.tanks[src].lnnode.connect_to_tank(dst)
+
+        # Start confirming channel opens in block 300
+        self.log.info("Opening channels, one per block")
+        chan_opens = []
+        edges = self.warnet.graph.edges(data=True, keys=True)
+        edges = sorted(edges, key=lambda edge: edge[2])
+        for edge in edges:
+            (src, dst, key, data) = edge
+            if "channel_open" in data:
                 src_node = self.warnet.get_ln_node_from_tank(src)
                 assert src_node is not None
                 assert self.warnet.get_ln_node_from_tank(dst) is not None
-                ln_edges.append(edge)
-                tx = src_node.open_channel_to_tank(dst, data["channel"])["funding_txid"]
-                opening_txs.append(tx)
+                self.log.info(f"opening channel {src}->{dst}")
+                chan_pt = src_node.open_channel_to_tank(dst, data["channel_open"])
+                # We can guarantee deterministic short channel IDs as long as
+                # the change output is greater than the channel funding output,
+                # which will then be output 0
+                assert chan_pt[64:] == ":0"
+                chan_opens.append((edge, chan_pt))
+                self.log.info(f"  pending channel point: {chan_pt}")
+                self.wait_until(lambda chan_pt=chan_pt: chan_pt[:64] in self.nodes[0].getrawmempool())
+                self.generatetoaddress(self.nodes[0], 1, miner_addr)
+                assert chan_pt[:64] not in self.nodes[0].getrawmempool()
+                height = self.nodes[0].getblockcount()
+                self.log.info(f"  confirmed in block {height}")
+                self.log.info(f"  channel_id should be: {int.from_bytes(height.to_bytes(3, 'big') + (1).to_bytes(3, 'big') + (0).to_bytes(2, 'big'), 'big')}")
 
-        self.log.info("Waiting for all channel open txs in mempool")
-        while True:
-            all_set = True
-            mp = self.nodes[3].getrawmempool()
-            for tx in opening_txs:
-                if tx not in mp:
-                    all_set = False
-            if all_set:
-                break
-            sleep(2)
+        # Ensure all channel opens are sufficiently confirmed
+        self.generatetoaddress(self.nodes[0], 10, miner_addr)
+        ln_nodes_gossip = ln_nodes.copy()
+        while len(ln_nodes_gossip) > 0:
+            self.log.info(f"Waiting for graph gossip sync, LN nodes remaining: {ln_nodes_gossip}")
+            for index in ln_nodes_gossip:
+                lnnode = self.warnet.tanks[index].lnnode
+                my_edges = len(lnnode.lncli("describegraph")["edges"])
+                my_nodes = len(lnnode.lncli("describegraph")["nodes"])
+                if my_edges == len(chan_opens) and my_nodes == len(ln_nodes):
+                    ln_nodes_gossip.remove(index)
+                else:
+                    self.log.info(f" node {index} not synced (channels: {my_edges}/{len(chan_opens)}, nodes: {my_nodes}/{len(ln_nodes)})")
+            sleep(5)
 
-        self.log.info("Confirming channel opens")
-        self.generatetoaddress(self.nodes[3], 6, addr)
+        self.log.info("Updating channel policies")
+        for edge, chan_pt in chan_opens:
+            (src, dst, key, data) = edge
+            if "target_policy" in data:
+                target_node = self.warnet.get_ln_node_from_tank(dst)
+                target_node.update_channel_policy(chan_pt, data["target_policy"])
+            if "source_policy" in data:
+                source_node = self.warnet.get_ln_node_from_tank(src)
+                source_node.update_channel_policy(chan_pt, data["source_policy"])
 
-        self.log.info("Waiting for graph gossip sync")
         while True:
-            all_set = True
-            for tank in self.warnet.tanks:
-                if tank.lnnode is not None:
-                    edges = tank.lnnode.lncli("describegraph")["edges"]
-                    if len(edges) != len(ln_edges):
-                        all_set = False
-            if all_set:
+            self.log.info("Waiting for all channel policies to match")
+            score = 0
+            for tank_index, me in enumerate(ln_nodes):
+                you = (tank_index + 1) % len(ln_nodes)
+                my_channels = self.warnet.tanks[me].lnnode.lncli("describegraph")["edges"]
+                your_channels = self.warnet.tanks[you].lnnode.lncli("describegraph")["edges"]
+                match = True
+                for chan_index, my_chan in enumerate(my_channels):
+                    your_chan = your_channels[chan_index]
+                    if not channel_match(my_chan, your_chan, allow_flip=False):
+                        print(f"Channel policy doesn't match between tanks {me} & {you}: {my_chan['channel_id']}")
+                        match = False
+                        break
+                if match:
+                    print(f"All channel policies match between tanks {me} & {you}")
+                    score += 1
+            print(f"Score: {score} / {len(ln_nodes)}")
+            if score == len(ln_nodes):
                 break
-            sleep(2)
-
-        self.log.info(f"Warnet LN ready with {len(recv_addrs)} nodes and {len(ln_edges)} channels.")
+            sleep(5)
 
+        self.log.info(f"Warnet LN ready with {len(recv_addrs)} nodes and {len(chan_opens)} channels.")
 
 if __name__ == "__main__":
     LNInit().main()

Diff for: src/warnet/utils.py

@@ -482,3 +482,25 @@ def validate_graph_schema(graph: nx.Graph):
         validate(instance=graph.nodes[n], schema=graph_schema["node"])
     for e in list(graph.edges):
         validate(instance=graph.edges[e], schema=graph_schema["edge"])
+
+
+def policy_match(pol1, pol2):
+    return (
+        pol1["time_lock_delta"] == pol2["time_lock_delta"]
+        and pol1["min_htlc"] == pol2["min_htlc"]
+        and pol1["fee_base_msat"] == pol2["fee_base_msat"]
+        and pol1["fee_rate_milli_msat"] == pol2["fee_rate_milli_msat"]
+        # Ignoring this for now since we use capacity/2
+        # and pol1["max_htlc_msat"] == pol2["max_htlc_msat"]
+    )
+
+
+def channel_match(ch1, ch2, allow_flip=False):
+    if ch1["capacity"] != ch2["capacity"]:
+        return False
+    if policy_match(ch1["node1_policy"], ch2["node1_policy"]) and policy_match(ch1["node2_policy"], ch2["node2_policy"]):
+        return True
+    if not allow_flip:
+        return False
+    else:
+        return policy_match(ch1["node1_policy"], ch2["node2_policy"]) and policy_match(ch1["node2_policy"], ch2["node1_policy"])