ln: save channels in dict, warn on invoice exceeding max_htlc amount

lib/lnbase.py

@@ -26,7 +26,7 @@ from cryptography.hazmat.backends import default_backend
 from .ecc import ser_to_point, point_to_ser, string_to_number
 from .bitcoin import (deserialize_privkey, rev_hex, int_to_hex,
                       push_script, script_num_to_hex,
-                      add_number_to_script, var_int)
+                      add_number_to_script, var_int, COIN)
 from . import bitcoin
 from . import ecc
 from . import crypto
@@ -36,6 +36,7 @@ from . import transaction
 from .util import PrintError, bh2u, print_error, bfh, profiler, xor_bytes
 from .transaction import opcodes, Transaction
 from .lnrouter import new_onion_packet, OnionHopsDataSingle, OnionPerHop
+from .lightning_payencode.lnaddr import lndecode
 
 from collections import namedtuple, defaultdict
 
@@ -586,7 +587,6 @@ class Peer(PrintError):
         self.update_fulfill_htlc = defaultdict(asyncio.Queue)
         self.commitment_signed = defaultdict(asyncio.Queue)
         self.localfeatures = (0x08 if request_initial_sync else 0)
-        self.unfulfilled_htlcs = []
         self.channel_state = channel_state
         self.nodes = {}
         self.channels = channels
@@ -769,7 +769,7 @@ class Peer(PrintError):
         msg = await self.read_message()
         self.process_message(msg)
         # reestablish channels
-        [await self.reestablish_channel(c) for c in self.channels]
+        [await self.reestablish_channel(c) for c in self.channels.values()]
         # loop
         while True:
             self.ping_if_required()
@@ -938,11 +938,11 @@ class Peer(PrintError):
 
     def on_channel_reestablish(self, payload):
         chan_id = int.from_bytes(payload["channel_id"], 'big')
-        for chan in self.channels:
+        for chan in self.channels.values():
             if chan.channel_id == chan_id:
                 break
         else:
-            print("Warning: received unknown channel_reestablish", chan_id, list(self.channels))
+            print("Warning: received unknown channel_reestablish", chan_id, list(self.channels.values()))
             return
         channel_reestablish_msg = payload
         remote_ctn = int.from_bytes(channel_reestablish_msg["next_local_commitment_number"], 'big')
@@ -1114,16 +1114,19 @@ class Peer(PrintError):
             )
         )
 
-    async def receive_commitment_revoke_ack(self, htlc, expected_received_msat, payment_preimage):
+    @aiosafe
-
+    async def receive_commitment_revoke_ack(self, htlc, decoded, payment_preimage):
-        htlc_id = int.from_bytes(htlc["id"], 'big')
+        chan = self.channels[int.from_bytes(htlc['channel_id'], 'big')]
-        for chan in self.channels:
-            if htlc_id == chan.remote_state.next_htlc_id:
-                break
-        else:
-            raise Exception('cannot find channel for htlc', htlc_id)
-
         channel_id = chan.channel_id
+        expected_received_msat = int(decoded.amount * COIN * 1000)
+        while True:
+            self.print_error("receiving commitment")
+            commitment_signed_msg = await self.commitment_signed[channel_id].get()
+            if int.from_bytes(commitment_signed_msg["num_htlcs"], "big") == 1:
+                break
+        htlc_id = int.from_bytes(htlc["id"], 'big')
+        assert htlc_id == chan.remote_state.next_htlc_id, (htlc_id, chan.remote_state.next_htlc_id)
+
         assert self.channel_state[channel_id] == "OPEN"
         their_revstore = chan.remote_state.revocation_store
         cltv_expiry = int.from_bytes(htlc["cltv_expiry"], 'big')
@@ -1216,7 +1219,7 @@ class Peer(PrintError):
             per_commitment_secret=last_secret,
             next_per_commitment_point=next_point))
 
-        return chan._replace(
+        new_chan = chan._replace(
             local_state=chan.local_state._replace(
                 amount_msat=chan.local_state.amount_msat + expected_received_msat
             ),
@@ -1229,6 +1232,7 @@ class Peer(PrintError):
                 next_htlc_id=htlc_id + 1
             )
         )
+        # TODO save new_chan
 
     def on_commitment_signed(self, payload):
         self.print_error("commitment_signed", payload)
@@ -1245,16 +1249,18 @@ class Peer(PrintError):
     def on_update_add_htlc(self, payload):
         # no onion routing for the moment: we assume we are the end node
         self.print_error('on_update_add_htlc', payload)
-        assert self.unfulfilled_htlcs == []
-        self.unfulfilled_htlcs.append(payload)
         # check if this in our list of requests
         payment_hash = payload["payment_hash"]
         for k in self.invoices.keys():
             preimage = bfh(k)
             if sha256(preimage) == payment_hash:
                 req = self.invoices[k]
-                coro = self.receive_commitment_revoke_ack(payload, chan, expected_received_msat, payment_preimage)
+                decoded = lndecode(req, expected_hrp=constants.net.SEGWIT_HRP)
-                future = asyncio.run_coroutine_threadsafe(coro, self.network.asyncio_loop)
+                coro = self.receive_commitment_revoke_ack(payload, decoded, preimage)
+                asyncio.run_coroutine_threadsafe(coro, self.network.asyncio_loop)
+                break
+        else:
+            assert False
 
     def on_revoke_and_ack(self, payload):
         channel_id = int.from_bytes(payload["channel_id"], 'big')
@@ -1297,4 +1303,6 @@ class RevocationStore:
         store.index = decoded_json_obj["index"]
         return store
     def __eq__(self, o):
-        return self.buckets == o.buckets and self.index == o.index
+        return type(o) is RevocationStore and self.serialize() == o.serialize()
+    def __hash__(self):
+        return hash(json.dumps(self.serialize(), sort_keys=True))

lib/lnworker.py

@@ -58,9 +58,9 @@ def reconstruct_namedtuples(openingchannel):
     openingchannel = openingchannel._replace(constraints=ChannelConstraints(**openingchannel.constraints))
     return openingchannel
 
-def serialize_channels(channels):
+def serialize_channels(channels_dict):
     serialized_channels = []
-    for chan in channels:
+    for chan in channels_dict.values():
         namedtuples_to_dict = lambda v: {i: j._asdict() if isinstance(j, tuple) else j for i, j in v._asdict().items()}
         serialized_channels.append({k: namedtuples_to_dict(v) if isinstance(v, tuple) else v for k, v in chan._asdict().items()})
     class MyJsonEncoder(json.JSONEncoder):
@@ -72,8 +72,8 @@ def serialize_channels(channels):
             return super(MyJsonEncoder, self)
     dumped = MyJsonEncoder().encode(serialized_channels)
     roundtripped = json.loads(dumped)
-    reconstructed = [reconstruct_namedtuples(x) for x in roundtripped]
+    reconstructed = set(reconstruct_namedtuples(x) for x in roundtripped)
-    if reconstructed != channels:
+    if reconstructed != set(channels_dict.values()):
         raise Exception("Channels did not roundtrip serialization without changes:\n" + repr(reconstructed) + "\n" + repr(channels))
     return roundtripped
 
@@ -85,8 +85,6 @@ node_list = [
     ('ecdsa.net', '9735', '038370f0e7a03eded3e1d41dc081084a87f0afa1c5b22090b4f3abb391eb15d8ff'),
 ]
 
-
-
 class LNWorker(PrintError):
 
     def __init__(self, wallet, network):
@@ -104,10 +102,10 @@ class LNWorker(PrintError):
         self.nodes = {}  # received node announcements
         self.channel_db = lnrouter.ChannelDB()
         self.path_finder = lnrouter.LNPathFinder(self.channel_db)
-        self.channels = [reconstruct_namedtuples(x) for x in wallet.storage.get("channels", {})]
+        self.channels = {x.channel_id: reconstruct_namedtuples(x) for x in wallet.storage.get("channels", {})}
         self.invoices = wallet.storage.get('lightning_invoices', {})
         peer_list = network.config.get('lightning_peers', node_list)
-        self.channel_state = {chan.channel_id: "OPENING" for chan in self.channels}
+        self.channel_state = {chan.channel_id: "OPENING" for chan in self.channels.values()}
         for host, port, pubkey in peer_list:
             self.add_peer(host, int(port), pubkey)
 
@@ -116,9 +114,13 @@ class LNWorker(PrintError):
         self.network.register_callback(self.on_network_update, ['updated', 'verified']) # thread safe
         self.on_network_update('updated') # shortcut (don't block) if funding tx locked and verified
 
+    def channels_for_peer(self, node_id):
+        assert type(node_id) is bytes
+        return {x: y for (x, y) in self.channels.items() if y.node_id == node_id}
+
     def add_peer(self, host, port, pubkey):
         node_id = bfh(pubkey)
-        channels = list(filter(lambda x: x.node_id == node_id, self.channels))
+        channels = self.channels_for_peer(node_id)
         peer = Peer(host, int(port), node_id, self.privkey, self.network, self.channel_db, self.path_finder, self.channel_state, channels, self.invoices, request_initial_sync=True)
         self.network.futures.append(asyncio.run_coroutine_threadsafe(peer.main_loop(), asyncio.get_event_loop()))
         self.peers[node_id] = peer
@@ -127,7 +129,9 @@ class LNWorker(PrintError):
     def save_channel(self, openchannel):
         if openchannel.channel_id not in self.channel_state:
             self.channel_state[openchannel.channel_id] = "OPENING"
-        self.channels = [openchannel] # TODO multiple channels
+        self.channels[openchannel.channel_id] = openchannel
+        for node_id, peer in self.peers.items():
+            peer.channels = self.channels_for_peer(node_id)
         dumped = serialize_channels(self.channels)
         self.wallet.storage.put("channels", dumped)
         self.wallet.storage.write()
@@ -154,7 +158,7 @@ class LNWorker(PrintError):
         return None
 
     def on_network_update(self, event, *args):
-        for chan in self.channels:
+        for chan in self.channels.values():
             if self.channel_state[chan.channel_id] == "OPEN":
                 continue
             chan = self.save_short_chan_id(chan)
@@ -184,7 +188,7 @@ class LNWorker(PrintError):
         self.on_channels_updated()
 
     def on_channels_updated(self):
-        std_chan = [{"chan_id": chan.channel_id} for chan in self.channels]
+        std_chan = [{"chan_id": chan.channel_id} for chan in self.channels.values()]
         self.trigger_callback('channels_updated', {'channels':std_chan})
 
     def open_channel(self, node_id, local_amt_sat, push_amt_sat, pw):
@@ -197,7 +201,7 @@ class LNWorker(PrintError):
 
     # not aiosafe because we call .result() which will propagate an exception
     async def _pay_coroutine(self, invoice):
-        openchannel = self.channels[0]
+        openchannel = next(iter(self.channels.values()))
         addr = lndecode(invoice, expected_hrp=constants.net.SEGWIT_HRP)
         payment_hash = addr.paymenthash
         pubkey = addr.pubkey.serialize()
@@ -207,10 +211,16 @@ class LNWorker(PrintError):
         self.save_channel(openchannel)
 
     def add_invoice(self, amount_sat, message='one cup of coffee'):
-        is_open = lambda chan: self.channel_state[chan] == "OPEN"
+        coro = self._add_invoice_coroutine(amount_sat, message)
+        return asyncio.run_coroutine_threadsafe(coro, self.network.asyncio_loop).result()
+
+    async def _add_invoice_coroutine(self, amount_sat, message):
+        is_open = lambda chan: self.channel_state[chan.channel_id] == "OPEN"
         # TODO doesn't account for fees!!!
-        #if not any(openchannel.remote_state.amount_msat >= amount_sat * 1000 for openchannel in self.channels if is_open(chan)):
+        if not any(openchannel.remote_state.amount_msat >= amount_sat * 1000 for openchannel in self.channels.values() if is_open(openchannel)):
-        #    return "Not making invoice, no channel has enough balance"
+            return "Not making invoice, no channel has enough balance"
+        if not any(openchannel.remote_config.max_htlc_value_in_flight_msat >= amount_sat * 1000 for openchannel in self.channels.values() if is_open(openchannel)):
+            return "Not making invoice, invoice value too lang for remote peer"
         payment_preimage = os.urandom(32)
         RHASH = sha256(payment_preimage)
         pay_req = lnencode(LnAddr(RHASH, amount_sat/Decimal(COIN), tags=[('d', message)]), self.privkey)