diff --git a/electrum/lnpeer.py b/electrum/lnpeer.py index 4b88f87d6e7..e972fa1a2e8 100644 --- a/electrum/lnpeer.py +++ b/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,6 +172,7 @@ 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() @@ -178,6 +180,10 @@ def is_initialized(self) -> bool: 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) diff --git a/electrum/lnworker.py b/electrum/lnworker.py index 874c74987f8..cc08350dadf 100644 --- a/electrum/lnworker.py +++ b/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) diff --git a/tests/test_lnpeer.py b/tests/test_lnpeer.py index 19e02311552..ee515cab6bf 100644 --- a/tests/test_lnpeer.py +++ b/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): diff --git a/tests/test_lnpeermgr.py b/tests/test_lnpeermgr.py index eccbda8da19..8c7f67e60c7 100644 --- a/tests/test_lnpeermgr.py +++ b/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