GlcoinCore/lndmanage
1
0
Fork 0
mirror of https://github.com/bitromortac/lndmanage.git synced 2026-08-13 12:33:37 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

routing: refactor route construction with rpc route

Browse source 
Removes legacy route construction and uses the lnd rpc instead. We add a
method to check the route and to add debug info.
This commit is contained in:
bitromortac 2022-10-01 19:10:43 +02:00
parent ef99b60051
commit decff1618a
No known key found for this signature in database
GPG key ID: 1965063FC13BEBE2
2 changed files with 188 additions and 202 deletions
Download patch file Download diff file Expand all files Collapse all files

5
lndmanage/lib/pathfinding.py
View file

@ -1,4 +1,5 @@
from typing import Callable, List
from lndmanage.lib.exceptions import NoRoute
import networkx as nx
@ -22,4 +23,8 @@ def dijkstra(graph: nx.Graph, source: str, target: str, weight: Callable) -> Lis
:return: hops in terms of the node keys
"""
path = nx.shortest_path(graph, source, target, weight=weight)
if not path:
raise NoRoute
return path

385
lndmanage/lib/routing.py
View file

@ -1,283 +1,264 @@
from typing import List, Dict, TYPE_CHECKING
import random
import grpc
from lndmanage.lib.data_types import NodePair
from lndmanage.lib.exceptions import RouteWithTooSmallCapacity, NoRoute
from lndmanage.lib.exceptions import NoRoute
from lndmanage.lib.pathfinding import dijkstra
from lndmanage import settings
if TYPE_CHECKING:
from lndmanage.lib.node import LndNode
import lndmanage.grpc_compiled.lightning_pb2 as lnd
import logging
logger = logging.getLogger(__name__)
logger.addHandler(logging.NullHandler())
def calculate_fees_on_policy(amt_msat, policy):
return policy['fee_base_msat'] + amt_msat * policy['fee_rate_milli_msat'] // 1000000
class Route(object):
"""Deals with the onion route construction from list of channels. Calculates fees
and cltvs.
"""
def __init__(self, node: 'LndNode', channel_hops: List[int], node_dest: str, amt_msat: int):
"""
:param node: :class:`lib.node.Node` instance
:param channel_hops: list of chan_ids along which the route shall be constructed
:param node_dest: pub_key of destination node
:param amt_msat: amount to send in msat
"""
self.node = node
self.blockheight = node.blockheight
logger.debug(f"Blockheight: {self.blockheight}")
self.channel_hops = channel_hops
self._hops = []
self._node_hops = [node_dest]
forward_msat = amt_msat
final_cltv = 144
fees_msat_container = [0]
cltv_delta = [0]
node_to = node_dest
node_from = None
policy = None
logger.debug("Route construction starting.")
# hops are traversed in backwards direction to accumulate fees and cltvs
for ichannel, channel_id in enumerate(reversed(channel_hops)):
channel_data = self.node.network.edges[channel_id]
# TODO: add private channels for sending
if amt_msat // 1000 > channel_data['capacity']:
logger.debug(f"Discovered a channel {channel_id} with too small capacity.")
raise RouteWithTooSmallCapacity(f"Amount too large for channel.")
policies = channel_data['policies']
if node_to == channel_data['node2_pub']:
try:
policy = policies[channel_data['node1_pub'] > channel_data['node2_pub']]
node_from = channel_data['node1_pub']
except KeyError:
logger.exception(f"No channel {channel_data}")
else:
policy = policies[channel_data['node2_pub'] > channel_data['node1_pub']]
node_from = channel_data['node2_pub']
self._node_hops.append(node_from)
hop = len(channel_hops) - ichannel
logger.info(f" Hop {hop}: {channel_id} (cap: {channel_data['capacity']} sat): "
f"{self.node.network.node_alias(node_to)} <- {self.node.network.node_alias(node_from)} ")
logger.debug(f" Policy of forwarding node: {policy}")
fees_msat = policy['fee_base_msat'] + policy['fee_rate_milli_msat'] * forward_msat // 1000000
forward_msat = amt_msat + sum(fees_msat_container[:ichannel])
fees_msat_container.append(fees_msat)
node_pair = NodePair((channel_data['node1_pub'], channel_data['node2_pub']))
capacity = self.node.network.max_pair_capacity[node_pair]
logger.info(f" Fees: {fees_msat / 1000 if not hop == 1 else 0:3.3f} sat")
logger.debug(f" Fees container {fees_msat_container}")
logger.debug(f" Forward: {forward_msat / 1000:3.3f} sat")
logger.info(f" Liquidity penalty: {self.node.network.liquidity_hints.penalty(node_from, node_to, capacity, amt_msat, self.node.network.channel_rater.reference_fee_rate_milli_msat) / 1000: 3.3f} sat")
logger.info(f" Badness penalty: {self.node.network.liquidity_hints.badness_penalty(node_from, amt_msat) / 1000: 3.3f} sat")
logger.info(f" Time penalty: {self.node.network.liquidity_hints.time_penalty(node_from, amt_msat) / 1000: 3.3f} sat")
self._hops.append({
'chan_id': channel_data['channel_id'],
'chan_capacity': channel_data['capacity'],
'amt_to_forward': forward_msat // 1000,
'fee': fees_msat_container[-2] // 1000,
'expiry': self.blockheight + final_cltv + sum(cltv_delta[:ichannel]),
'amt_to_forward_msat': forward_msat,
'fee_msat': fees_msat_container[-2],
})
cltv_delta.append(policy['time_lock_delta'])
node_to = node_from
self.hops = list(reversed(self._hops))
self.node_hops = list(reversed(self._node_hops))
self.total_amt_msat = amt_msat + sum(fees_msat_container[:-1])
self.total_fee_msat = sum(fees_msat_container[:-1])
self.total_time_lock = sum(cltv_delta[:-1]) + self.blockheight + final_cltv
def _debug_route(self):
"""Prints detailed information of the route."""
logger.debug("Debug route:")
for h in self.hops:
logger.debug(f"c:{h['chan_id']} a:{h['amt_to_forward']} f:{h['fee_msat']}"
f" c:{h['expiry'] - self.node.blockheight}")
channel_info = self.node.network.edges[h['chan_id']]
node1 = channel_info['node1_pub']
node2 = channel_info['node2_pub']
logger.debug(f"{node1[:5]}, {channel_info['node1_policy']}")
logger.debug(f"{node2[:5]}, {channel_info['node2_policy']}")
logger.debug(f"tl:{self.total_time_lock} ta:{self.total_amt_msat} tf:{self.total_fee_msat}")
def fees_for_policy(amt_msat, policy):
return policy['fee_base_msat'] + \
amt_msat * policy['fee_rate_milli_msat'] // 1000000
class Router(object):
"""Contains utilities for constructing routes."""
"""Contains utilities for route construction."""
def __init__(self, node: 'LndNode'):
self.node = node
def _node_route_to_channel_route(self, node_route: List[str], amt_msat: int) -> List[int]:
"""Takes a route in terms of a list of nodes and translates it into a list of
channels.
:param node_route: list of pubkeys
:param amt_msat: amount to send in sat
:return: list of channel_ids
"""
channels = []
for p in range(len(node_route) - 1):
channels.append(
self._determine_channel(
node_route[p], node_route[p + 1], amt_msat)[1])
return channels
def get_route_from_to_nodes(self, node_from: str, node_to: str, amt_msat: int) -> List[str]:
"""Determines number_of_routes shortest paths between node_from and node_to for
an amount of amt_msat.
def find_path(self, node_from: str, node_to: str,
amt_msat: int) -> List[str]:
"""Looks for a path from one node to another for a certain amount.
:param node_from: pubkey
:param node_to: pubkey
:param amt_msat: amount to send in msat
:return: route
:return: list of pubkey hops
"""
self.node.network.channel_rater.blacklisted_nodes.append(self.node.pub_key) # excludes self-loops
weight_function = lambda v, u, e: self.node.network.channel_rater.node_to_node_weight(v, u, e, amt_msat)
route = dijkstra(self.node.network.graph, node_from, node_to, weight=weight_function)
# Exclude self-loops.
self.node.network.channel_rater.blacklisted_nodes.append(
self.node.pub_key,
)
if not route:
raise NoRoute
def weight_function(v, u, e):
return self.node.network.channel_rater.node_to_node_weight(
v, u, e, amt_msat,
)
# Perform a Dijkstra shortest path search.
# TODO: known limitation: does not include fees of fees.
route = dijkstra(
self.node.network.graph, node_from, node_to, weight=weight_function,
)
return route
def _determine_channel(self, node_from: str, node_to: str, amt_msat: int):
"""Determines the cheapest channel between nodes node_from and node_to for an
amount of amt_msat.
def check_route(self, route):
"""Checks a route for sanity and gives debug output."""
:param node_from: pubkey
:param node_to: pubkey
:param amt_msat: amount to send in msat
:return: channel_id
"""
number_edges = self.node.network.graph.number_of_edges(node_from, node_to)
channels_with_calculated_fees = []
for n in range(number_edges):
edge = self.node.network.graph.get_edge_data(node_from, node_to, n)
fees = self.node.network.channel_rater.channel_weight(node_from, node_to, edge, amt_msat)
channels_with_calculated_fees.append([fees, edge['channel_id']])
sorted_channels = sorted(channels_with_calculated_fees, key=lambda k: k[0])
best_channel = sorted_channels[0]
# we check that we don't encounter a hop which is blacklisted
if best_channel[0] == float('inf'):
raise NoRoute('channels graph exhausted')
return best_channel
# We check that the route is not just sending and receiveing over the
# same channel.
if len(route.hops) == 2:
raise NoRoute(f"only minimal route available: self -> other -> "+
"self: {rpc_error.details()}")
def _determine_cheapest_fees_between_two_nodes(self, node_from, node_to, amt_msat):
return self._determine_channel(node_from, node_to, amt_msat)[0]
# We check that the chosen channels have some finite chance of success
# and that they are not blacklisted.
node_from = self.node.pub_key
for i, hop in enumerate(route.hops):
node_to = hop.pub_key
# We don't want our node to be inside the path.
if i > 0 and i < len(route.hops) - 1:
assert node_to != self.node.pub_key, "our node is inside of the path"
# Fetch the channel policy.
edge_data = None
edges = self.node.network.graph[node_from][node_to]
for edge in edges.values():
if edge['channel_id'] == hop.chan_id:
edge_data = edge
break
if not edge_data:
raise NoRoute("channel not found in local graph")
# Display some debug output.
logger.info(f" Hop {i}: {hop.chan_id} (cap: {edge_data['capacity']} sat): "
f"{self.node.network.node_alias(node_from)} -> " +
f"{self.node.network.node_alias(node_to)}")
logger.debug(f" Fees next: {hop.fee_msat:9.3f} sat")
_ = self.node.network.channel_rater.channel_weight(
node_from, node_to, edge_data, hop.amt_to_forward_msat,
)
node_from = node_to
def find_node_route(self, source_pubkey: str, target_pubkey: str,
amt_msat: int) -> List[int]:
"""Finds a route from source to target for a certain amount. Returns a
list of pubkeys."""
def get_route_channel_hops_from_to_node_internal(
self,
source_pubkey: str,
target_pubkey: str,
amt_msat: int
) -> List[int]:
"""Find routes internally, using networkx to construct a route from a source
node to a target node."""
logger.debug(f"Internal pathfinding:")
logger.debug(f"from {source_pubkey}")
logger.debug(f" to {target_pubkey}")
node_route = self.get_route_from_to_nodes(
node_hops = self.find_path(
source_pubkey, target_pubkey, amt_msat)
return self._node_route_to_channel_route(node_route, amt_msat)
return node_hops
def get_route(
def route_from_constraints(
self,
send_channels: Dict[int, dict],
receive_channels: Dict[int, dict],
amt_msat: int
) -> Route:
"""Calculates a route from send_channels to receive_channels.
amt_msat: int,
payment_addr: bytes,
) -> 'lnd.Route':
"""Calculates a route that leaves over send_channels and enters via
receive_channels.
:param send_channels: channel ids to send from
:param receive_channels: channel ids to receive to
:param amt_msat: payment amount in msat
:return: a route for rebalancing
:return: a route that can be sent do via the lnd api
"""
this_node = self.node.pub_key # TODO: make this a parameter for general route calculation
this_node = self.node.pub_key
# Reset old blacklists.
self.node.network.channel_rater.reset_channel_blacklist()
# We will ask for a route from source to target.
# we send via a send channel and receive over other channels:
# this_node -(send channel)-> source -> ... -> receiver neighbors -(receive channels)-> target (this_node)
# We will ask for a route from source to target. The specific source and
# target depends on the input channels.
# Case1: single send channel and multiple receive channels:
# * this_node -(send channel)->
# [* source ->
# ... ->
# * receiver neighbors -(receive channels)->
# * target (this_node)]
# Look for a path in parantheses.
if len(send_channels) == 1:
# There is only a single send channel.
send_channel = list(send_channels.values())[0]
source = send_channel['remote_pubkey']
target = this_node
# we don't want to go backwards via the send_channel (and other parallel channels)
channels_source_target = self.node.network.graph[source][target]
for channel in channels_source_target.values():
self.node.network.channel_rater.blacklist_add_channel(channel['channel_id'], source, target)
# We don't want to go backwards via the send_channel and other
# parallel channels between source and target.
blocked_channels = self.node.network.graph[source][target]
for channel in blocked_channels.values():
self.node.network.channel_rater.blacklist_add_channel(
channel['channel_id'], source, target,
)
# We exclude all other channels other than receive channels from
# receiving.
# we want to use the receive channels for receiving only, so don't receive over other channels
excluded_receive_channels = self.node.get_unbalanced_channels(
excluded_channels=[k for k in receive_channels.keys()], public_only=False, active_only=False)
excluded_channels=[k for k in receive_channels.keys()],
public_only=False, active_only=False,
)
for channel_id, channel in excluded_receive_channels.items():
receiver_neighbor = channel['remote_pubkey']
self.node.network.channel_rater.blacklist_add_channel(channel_id, receiver_neighbor, target)
self.node.network.channel_rater.blacklist_add_channel(
channel_id, receiver_neighbor, target,
)
# we send via several channels and receive over a single one:
# this_node (source) -(send channels)-> ... -> receiver neighbor (target) -(receive channel)-> this_node
# Case 2: send via several channels and receive over a single one
# * [this_node (source) -(send channels)->
# * ... ->
# * receiver neighbor (target)] -(receive channel)->
# * this_node
elif len(receive_channels) == 1:
# We have only a single receive channel.
receive_channel = list(receive_channels.values())[0]
source = this_node
target = receive_channel['remote_pubkey']
# we want to block the receiving channel (and parallel ones) from sending
channels_source_target = self.node.network.graph[source][target]
for channel in channels_source_target.values():
self.node.network.channel_rater.blacklist_add_channel(channel['channel_id'], source, target)
# We want to block the receiving channel and parallel ones from
# sending.
blocked_channels = self.node.network.graph[source][target]
for channel in blocked_channels.values():
self.node.network.channel_rater.blacklist_add_channel(
channel['channel_id'], source, target,
)
# we want to use the send channels for sending only, so don't send over other channels
# We want to use the send channels for sending only, so don't send
# over other channels.
excluded_send_channels = self.node.get_unbalanced_channels(
excluded_channels=[k for k in send_channels.keys()], public_only=False, active_only=False)
excluded_channels=[k for k in send_channels.keys()],
public_only=False, active_only=False)
for channel_id, channel in excluded_send_channels.items():
sender_neighbor = channel['remote_pubkey']
self.node.network.channel_rater.blacklist_add_channel(channel_id, source, sender_neighbor)
self.node.network.channel_rater.blacklist_add_channel(
channel_id, source, sender_neighbor,
)
else:
raise ValueError("One of the two channel sets should be singular.")
# determine inner channel hops
# internal method uses networkx dijkstra,
# this is more independent, but slower
route_channel_hops = \
self.get_route_channel_hops_from_to_node_internal(
source, target, amt_msat)
# Up to this point, we have determined source and target channels.
final_channel_hops = []
# Compute hops from source to target.
hop_pubkeys = self.find_node_route(
source, target, amt_msat,
)
# Construct the final list of nodes.
final_hop_pubkeys = []
outgoing_channel = None
# Single-send channel, multiple receive channels.
if len(send_channels) == 1:
final_channel_hops.append(send_channel['chan_id'])
final_channel_hops.extend(route_channel_hops)
final_hop_pubkeys.extend(hop_pubkeys)
outgoing_channel = send_channel['chan_id']
# Multiple send channels, single receive channel.
else:
final_channel_hops.extend(route_channel_hops)
final_channel_hops.append(receive_channel['chan_id'])
# We need to extend the path with the pubkey of this node.
final_hop_pubkeys.extend(hop_pubkeys[1:])
final_hop_pubkeys.append(this_node)
# TODO: add some consistency checks, route shouldn't contain self-loops
logger.debug("Channel hops:")
logger.debug(final_channel_hops)
# For the outgoing channel, select a channel from the send channels
# with the nearest neighbor (second pubkey).
send_candidates = [c for c, v in send_channels.items() if
v['remote_pubkey'] == final_hop_pubkeys[0]]
# initialize Route objects with appropriate fees and expiries
route = Route(self.node, final_channel_hops, this_node, amt_msat)
# TODO: select a better send channel if multiple are available.
outgoing_channel = random.choice(send_candidates)
logger.debug("Node hops:")
logger.debug(final_hop_pubkeys)
logger.info(f"Construct route for {len(final_hop_pubkeys)} hops.")
# Build a route via an RPC call to LND.
try:
route = self.node.build_route(
amt_msat, outgoing_channel, final_hop_pubkeys, payment_addr,
)
except grpc.RpcError as rpc_error:
if rpc_error.code() == grpc.StatusCode.UNKNOWN:
if "for node 0" in rpc_error.details():
raise NoRoute(f"our node can't send: {rpc_error.details()}")
raise NoRoute(
f"could not build a route {final_hop_pubkeys}"
f"outgoing {outgoing_channel}: {rpc_error.details()}"
)
else:
raise rpc_error
self.check_route(route)
return route