lnpeermgr: keep peer connections bounded

electrum/lnpeer.py

@@ -49,7 +49,7 @@
                      IncompatibleOrInsaneFeatures, ReceivedMPPStatus, ReceivedMPPHtlc,
                      GossipForwardingMessage, GossipTimestampFilter, channel_id_from_funding_tx,
                      serialize_htlc_key, Keypair, RecvMPPResolution)
-from .lntransport import LNTransport, LNTransportBase, LightningPeerConnectionClosed, HandshakeFailed
+from .lntransport import LNTransport, LNTransportBase, LNResponderTransport, LightningPeerConnectionClosed, HandshakeFailed
 from .lnmsg import encode_msg, decode_msg, UnknownOptionalMsgType, FailedToParseMsg
 from .interface import GracefulDisconnect
 from .invoices import PR_PAID
@@ -110,6 +110,7 @@ def __init__(
         self.node_ids = [self.pubkey, privkey_to_pubkey(self.privkey)]
         assert self.node_ids[0] != self.node_ids[1]
         self.last_message_time = 0
+        self.initialization_time = None  # type: Optional[float]
         self.pong_event = asyncio.Event()
         self.reply_channel_range = None  # type: Optional[asyncio.Queue]
         # gossip uses a single queue to preserve message order
@@ -171,13 +172,18 @@ def maybe_set_initialized(self):
             return
         if self._sent_init and self._received_init:
             self.initialized.set_result(True)
+            self.initialization_time = time.monotonic()
 
     def is_initialized(self) -> bool:
         return (self.initialized.done()
                 and not self.initialized.cancelled()
                 and self.initialized.exception() is None
                 and self.initialized.result() is True)
 
+    def is_incoming(self) -> bool:
+        """Whether the remote party initiated the connection."""
+        return isinstance(self.transport, LNResponderTransport)
+
     async def initialize(self):
         # If outgoing transport, do handshake now. For incoming, it has already been done.
         if isinstance(self.transport, LNTransport):
@@ -892,6 +898,8 @@ async def _message_loop(self):
         try:
             await util.wait_for2(self.initialize(), LN_P2P_NETWORK_TIMEOUT)
         except (OSError, asyncio.TimeoutError, HandshakeFailed) as e:
+            if not self.initialized.done():
+                self.initialized.set_exception(e)  # forward exc so waiters can fail too
             raise GracefulDisconnect(f'initialize failed: {repr(e)}') from e
         async for msg in self.transport.read_messages():
             await self._process_message(msg)

electrum/lnworker.py

@@ -222,6 +222,8 @@ class ErrorAddingPeer(Exception): pass
 
 
 class LNPeerManager(Logger, EventListener, NetworkRetryManager[LNPeerAddr]):
+    MAX_CHANNELLESS_PEERS_PER_DIRECTION = 100
+    CHANNELLESS_PEER_GRACE_PERIOD = 60  # duration before we start disconnecting, to give time for actual action
 
     def __init__(
         self, node_keypair,
@@ -341,6 +343,50 @@ async def _maintain_connectivity(self):
                     except ErrorAddingPeer as e:
                         self.logger.info(f"failed to add peer: {peer}. exc: {e!r}")
 
+    def _get_channelless_peers(self) -> tuple[list[Peer], list[Peer]]:
+        """Returns all connected peers with which we have/had no channels yet, sorted by initialization timestamp."""
+        channels = self._lnwallet_or_lngossip.channels if isinstance(self._lnwallet_or_lngossip, LNWallet) else {}
+        # redeemed channels count (they cost sats to fake), unfunded channels need to be rate-limited separately
+        nodes_with_channels = {chan.node_id for chan in channels.values()}
+        incoming, outgoing = [], []
+        with self.lock:
+            for node_id, peer in self._peers.items():
+                if node_id not in nodes_with_channels and peer.initialization_time is not None:
+                    if peer.is_incoming():
+                        incoming.append(peer)
+                    else:
+                        outgoing.append(peer)
+        eviction_order = lambda p: p.initialization_time
+        incoming.sort(key=eviction_order, reverse=True)
+        outgoing.sort(key=eviction_order, reverse=True)
+        return incoming, outgoing
+
+    async def _cleanup_unused_peers(self):
+        """
+        Peers without channels can accumulate over time through manual connections, failed channel opening flows,
+        onion message fallbacks and incoming connections.
+        This regularly closes the oldest channelless connections once the amount of channelless peers
+        exceeds a threshold to keep the resource usage bounded.
+        """
+        assert isinstance(self._lnwallet_or_lngossip, LNWallet), "lngossip does _maintain_connectivity"
+        while True:
+            await asyncio.sleep(10)
+            channelless_peers = self._get_channelless_peers()
+            protected_peers = set(cp[2].lower() for cp in self.config.LIGHTNING_PEERS or [] if len(cp) == 3)  # host, port, pubkey
+            if self._lnwallet_or_lngossip.trusted_zeroconf_node_id:
+                protected_peers.add(self._lnwallet_or_lngossip.trusted_zeroconf_node_id.hex())
+            # incoming peers should only be able to exceed peer count limit by becoming channelless after
+            # connection. Incoming connections that would already exceed the count will get rejected right away.
+            for peers in channelless_peers:  # incoming, outgoing
+                while len(peers) > self.MAX_CHANNELLESS_PEERS_PER_DIRECTION:
+                    oldest_peer = peers.pop()
+                    if oldest_peer.pubkey.hex() in protected_peers:
+                        continue
+                    peer_age = time.monotonic() - oldest_peer.initialization_time
+                    if peer_age <= self.CHANNELLESS_PEER_GRACE_PERIOD:
+                        break
+                    oldest_peer.close_and_cleanup()
+
     async def _add_peer(self, host: str, port: int, node_id: bytes) -> Peer:
         if node_id in self._peers:
             return self._peers[node_id]
@@ -383,7 +429,7 @@ async def _add_peer_from_transport(self, *, node_id: bytes, transport: LNTranspo
                 assert node_id not in self._channelless_incoming_peers
                 chans = [chan for chan in self.channels_for_peer(node_id).values() if chan.is_funded()]
                 if not chans:
-                    if len(self._channelless_incoming_peers) > 100:
+                    if len(self._channelless_incoming_peers) >= self.MAX_CHANNELLESS_PEERS_PER_DIRECTION:
                         transport.close()
                         return None
                     self._channelless_incoming_peers.add(node_id)
@@ -432,6 +478,9 @@ def start_network(
         if maintain_random_peers:
             tg_coro = self.taskgroup.spawn(self._maintain_connectivity())
             asyncio.run_coroutine_threadsafe(tg_coro, get_asyncio_loop())
+        else:
+            cleanup_coro = self.taskgroup.spawn(self._cleanup_unused_peers())
+            asyncio.run_coroutine_threadsafe(cleanup_coro, get_asyncio_loop())
 
     async def stop(self):
         self.stopping_soon = True
@@ -580,6 +629,13 @@ def on_event_proxy_set(self, *args):
             peer.close_and_cleanup()
         self._clear_addr_retry_times()
 
+    @event_listener
+    def on_event_channel(self, wallet: 'Abstract_Wallet', channel: 'Channel'):
+        if channel.is_funded() \
+                and isinstance(self._lnwallet_or_lngossip, LNWallet) and self._lnwallet_or_lngossip.wallet == wallet:
+            with self.lock:
+                self._channelless_incoming_peers.discard(channel.node_id)
+
     @log_exceptions
     async def add_peer(self, connect_str: str) -> Peer:
         node_id, rest = extract_nodeid(connect_str)

tests/test_lnpeer.py

@@ -2,6 +2,7 @@
 import dataclasses
 import shutil
 import copy
+import socket
 import tempfile
 from decimal import Decimal
 import os
@@ -33,8 +34,8 @@
 from electrum.bitcoin import COIN, sha256
 from electrum.transaction import Transaction
 from electrum.util import NetworkRetryManager, bfh, OldTaskGroup, EventListener, InvoiceError
-from electrum.lnpeer import Peer
-from electrum.lntransport import LNPeerAddr
+from electrum.lnpeer import Peer, LN_P2P_NETWORK_TIMEOUT
+from electrum.lntransport import LNPeerAddr, LNTransport
 from electrum.crypto import privkey_to_pubkey
 from electrum.lnutil import Keypair, PaymentFailure, LnFeatures, HTLCOwner, PaymentFeeBudget, RECEIVED
 from electrum.lnchannel import ChannelState, PeerState, Channel
@@ -412,6 +413,28 @@ async def test_zeroconf_feature_bit(self):
             alice, _ = graph.peers.values()
             self.assertTrue(alice.features.supports(LnFeatures.OPTION_ZEROCONF_OPT))
 
+    async def test_initialized_future_resolves_on_failed_connection(self):
+        """The self.initialized future should return a connection exception if the connection fails, so consumers
+        awaiting it don't wait until their timeout if the connection failed"""
+        lnworker = self.create_mock_lnwallet(name='alice')
+        with socket.socket() as s:  # allocate a localhost port with no listener
+            s.bind(('127.0.0.1', 0))
+            port = s.getsockname()[1]
+        remote_pubkey = ECPrivkey.generate_random_key().get_public_key_bytes()
+        peer_addr = LNPeerAddr('127.0.0.1', port, remote_pubkey)
+        transport = LNTransport(lnworker.node_keypair.privkey, peer_addr, e_proxy=None)
+        peer = Peer(lnworker, remote_pubkey, transport)
+        main_loop_task = asyncio.create_task(peer.main_loop())
+        try:
+            async with timeout_after(LN_P2P_NETWORK_TIMEOUT // 5):
+                with self.assertRaises(ConnectionRefusedError):
+                    await peer.initialized
+                await main_loop_task
+            self.assertTrue(peer.got_disconnected.is_set())
+        finally:
+            if not main_loop_task.done():
+                main_loop_task.cancel()
+
 
 class TestPeerDirect(TestPeer):
 

tests/test_lnpeermgr.py

@@ -2,11 +2,13 @@
 import os
 import socket
 import asyncio
+import time
 from unittest import mock
 
 from . import ElectrumTestCase
 
-from electrum.lntransport import ConnStringFormatError
+from electrum.lnpeer import Peer
+from electrum.lntransport import ConnStringFormatError, LNPeerAddr, LNResponderTransport, LNTransport
 from electrum.logging import console_stderr_handler
 
 
@@ -59,6 +61,65 @@ async def test_add_peer_conn_string_errors(self):
                 await peermgr.add_peer(bad_host_conn_str)
             self.assertIn("Hostname does not resolve", str(cm.exception))
 
+    def _add_channelless_peer(self, *, incoming: bool, initialization_time: float) -> Peer:
+        peermgr = self.lnpeermgr
+        pubkey = os.urandom(33)
+        if incoming:
+            transport = LNResponderTransport(peermgr.node_keypair.privkey, None, None)
+            transport._pubkey = pubkey  # normally set during the handshake
+        else:
+            peer_addr = LNPeerAddr('127.0.0.1', 9735, pubkey)
+            transport = LNTransport(peermgr.node_keypair.privkey, peer_addr, e_proxy=None)
+        peer = Peer(peermgr._lnwallet_or_lngossip, pubkey, transport)
+        peer.initialization_time = initialization_time
+        peermgr._peers[pubkey] = peer
+        return peer
+
+    async def _run_single_cleanup_iteration(self):
+        num_sleeps = 0
+
+        async def limited_sleep(delay):
+            nonlocal num_sleeps
+            num_sleeps += 1
+            if num_sleeps > 1:  # run exactly one iteration of the cleanup loop
+                raise asyncio.CancelledError
+
+        with mock.patch.object(asyncio, 'sleep', limited_sleep):
+            with self.assertRaises(asyncio.CancelledError):
+                await self.lnpeermgr._cleanup_unused_peers()
+
+    async def test_cleanup_unused_peers_evicts_per_direction(self):
+        """Incoming connection pressure must not cause eviction of outgoing connections,
+        as we opened those for a purpose (e.g. an onion message session)."""
+        peermgr = self.lnpeermgr
+        now = time.monotonic()
+        outgoing = [self._add_channelless_peer(incoming=False, initialization_time=now - 7200) for _ in range(2)]
+        incoming = [self._add_channelless_peer(incoming=True, initialization_time=now - 3600 - i)
+                    for i in range(peermgr.MAX_CHANNELLESS_PEERS_PER_DIRECTION + 3)]
+        num_excess_incoming = len(incoming) - peermgr.MAX_CHANNELLESS_PEERS_PER_DIRECTION
+        await self._run_single_cleanup_iteration()
+        for peer in outgoing:  # outgoing peers survive even though they are the oldest
+            self.assertIn(peer.pubkey, peermgr.peers)
+        self.assertEqual(peermgr.MAX_CHANNELLESS_PEERS_PER_DIRECTION + len(outgoing), len(peermgr.peers))
+        # the incoming excess got evicted from the incoming peers, oldest first
+        evicted = [peer for peer in incoming if peer.pubkey not in peermgr.peers]
+        oldest_incoming = sorted(incoming, key=lambda p: p.initialization_time)[:num_excess_incoming]
+        self.assertEqual({p.pubkey for p in oldest_incoming}, {p.pubkey for p in evicted})
+
+    async def test_cleanup_unused_peers_keeps_young_outgoing_peers(self):
+        """Outgoing connections younger than the minimum lifetime are not evicted, even above the cap."""
+        peermgr = self.lnpeermgr
+        now = time.monotonic()
+        old = [self._add_channelless_peer(incoming=False, initialization_time=now - 3600) for _ in range(2)]
+        young = [self._add_channelless_peer(incoming=False, initialization_time=now - 5)
+                 for _ in range(peermgr.MAX_CHANNELLESS_PEERS_PER_DIRECTION + 2)]
+        await self._run_single_cleanup_iteration()
+        self.assertEqual(len(young), len(peermgr.peers))
+        for peer in old:
+            self.assertNotIn(peer.pubkey, peermgr.peers)
+        for peer in young:
+            self.assertIn(peer.pubkey, peermgr.peers)
+
     def test_choose_preferred_address(self):
         peermgr = self.lnpeermgr