chia-blockchain/chia/server/node_discovery.py

from __future__ import annotations

import asyncio
import contextlib
import math
import random
import time
import traceback
from collections.abc import AsyncIterator
from dataclasses import dataclass, field
from logging import Logger
from pathlib import Path
from random import Random
from typing import Any

import dns.asyncresolver
from chia_rs.sized_ints import uint16, uint64

from chia.protocols.full_node_protocol import RequestPeers, RespondPeers
from chia.protocols.introducer_protocol import RequestPeersIntroducer
from chia.protocols.outbound_message import Message, NodeType, make_msg
from chia.protocols.protocol_message_types import ProtocolMessageTypes
from chia.server.address_manager import AddressManager, ExtendedPeerInfo
from chia.server.server import ChiaServer
from chia.server.ws_connection import WSChiaConnection
from chia.types.peer_info import PeerInfo, TimestampedPeerInfo, UnresolvedPeerInfo
from chia.util.files import write_file_async
from chia.util.hash import std_hash
from chia.util.ip_address import IPAddress
from chia.util.network import resolve
from chia.util.safe_cancel_task import cancel_task_safe
from chia.util.task_referencer import create_referenced_task

MAX_PEERS_RECEIVED_PER_REQUEST = 1000
MAX_TOTAL_PEERS_RECEIVED = 3000
MAX_CONCURRENT_OUTBOUND_CONNECTIONS = 70
MAX_IP_ADDRESS_STRING_LENGTH = 45  # IPv4 max 15, IPv6 max 45
NETWORK_ID_DEFAULT_PORTS = {
    "mainnet": 8444,
    "testnet7": 58444,
    "testnet10": 58444,
    "testnet8": 58445,
}


@dataclass
class FullNodeDiscovery:
    server: ChiaServer
    target_outbound_count: int
    peers_file_path: Path
    dns_servers: list[str]
    peer_connect_interval: int
    selected_network: str
    log: Logger
    introducer_info: dict[str, Any] | None = None
    default_port: int | None = None
    resolver: dns.asyncresolver.Resolver | None = field(default=None)
    enable_private_networks: bool = field(default=False)
    is_closed: bool = field(default=False)
    legacy_peer_db_migrated: bool = field(default=False)
    relay_queue: asyncio.Queue[tuple[TimestampedPeerInfo, int]] | None = field(default=None)
    address_manager: AddressManager | None = field(default=None)
    connection_time_pretest: dict[str, Any] = field(default_factory=dict)
    received_count_from_peers: dict[str, Any] = field(default_factory=dict)
    lock: asyncio.Lock = field(default_factory=asyncio.Lock)
    connect_peers_task: asyncio.Task[None] | None = field(default=None)
    serialize_task: asyncio.Task[None] | None = field(default=None)
    cleanup_task: asyncio.Task[None] | None = field(default=None)
    initial_wait: int = field(default=0)
    pending_outbound_connections: set[str] = field(default_factory=set)
    pending_tasks: set[asyncio.Task[None]] = field(default_factory=set)
    introducer_info_obj: UnresolvedPeerInfo | None = field(default=None)

    def __post_init__(self) -> None:
        random.shuffle(self.dns_servers)  # Don't always start with the same DNS server

        if self.introducer_info is not None:
            self.introducer_info_obj = UnresolvedPeerInfo(self.introducer_info["host"], self.introducer_info["port"])
            self.enable_private_networks = self.introducer_info.get("enable_private_networks", False)

        try:
            self.resolver = dns.asyncresolver.Resolver()
        except Exception:
            self.resolver = None
            self.log.exception("Error initializing asyncresolver")

        if self.default_port is None and self.selected_network in NETWORK_ID_DEFAULT_PORTS:
            self.default_port = NETWORK_ID_DEFAULT_PORTS[self.selected_network]

    async def initialize_address_manager(self) -> None:
        self.address_manager = await AddressManager.create_address_manager(self.peers_file_path)
        if self.enable_private_networks:
            self.address_manager.make_private_subnets_valid()
        self.server.set_received_message_callback(self.update_peer_timestamp_on_message)

    async def start_tasks(self) -> None:
        random = Random()
        self.connect_peers_task = create_referenced_task(self._connect_to_peers(random))
        self.serialize_task = create_referenced_task(self._periodically_serialize(random))
        self.cleanup_task = create_referenced_task(self._periodically_cleanup())

    async def _close_common(self) -> None:
        self.is_closed = True
        cancel_task_safe(self.connect_peers_task, self.log)
        cancel_task_safe(self.serialize_task, self.log)
        cancel_task_safe(self.cleanup_task, self.log)
        tasks_to_await = {t for t in (self.connect_peers_task, self.serialize_task, self.cleanup_task) if t is not None}
        if len(tasks_to_await) > 0:
            await asyncio.wait(tasks_to_await)
        for t in self.pending_tasks:
            cancel_task_safe(t, self.log)
        if len(self.pending_tasks) > 0:
            await asyncio.wait(self.pending_tasks)

    async def on_connect(self, peer: WSChiaConnection) -> None:
        if (
            peer.is_outbound is False
            and peer.peer_server_port is not None
            and peer.connection_type is NodeType.FULL_NODE
            and self.server._local_type is NodeType.FULL_NODE
            and self.address_manager is not None
        ):
            timestamped_peer_info = TimestampedPeerInfo(
                peer.peer_info.host,
                peer.peer_server_port,
                uint64(time.time()),
            )
            await self.address_manager.add_to_new_table([timestamped_peer_info], peer.get_peer_info(), 0)
            if self.relay_queue is not None:
                self.relay_queue.put_nowait((timestamped_peer_info, 1))
        if (
            peer.is_outbound
            and peer.peer_server_port is not None
            and peer.connection_type is NodeType.FULL_NODE
            and (self.server._local_type is NodeType.FULL_NODE or self.server._local_type is NodeType.WALLET)
            and self.address_manager is not None
        ):
            msg = make_msg(ProtocolMessageTypes.request_peers, RequestPeers())
            await peer.send_message(msg)

    # Updates timestamps each time we receive a message for outbound connections.
    async def update_peer_timestamp_on_message(self, peer: WSChiaConnection) -> None:
        if (
            peer.is_outbound
            and peer.peer_server_port is not None
            and peer.connection_type is NodeType.FULL_NODE
            and self.server._local_type is NodeType.FULL_NODE
            and self.address_manager is not None
        ):
            peer_info = peer.get_peer_info()
            if peer_info is None:
                return None
            if peer_info.host not in self.connection_time_pretest:
                self.connection_time_pretest[peer_info.host] = time.time()
            if time.time() - self.connection_time_pretest[peer_info.host] > 600:
                self.connection_time_pretest[peer_info.host] = time.time()
                await self.address_manager.connect(peer_info)

    def _num_needed_peers(self) -> int:
        target = self.target_outbound_count
        outgoing = len(self.server.get_connections(NodeType.FULL_NODE, outbound=True))
        return max(0, target - outgoing)

    """
    Uses the Poisson distribution to determine the next time
    when we'll initiate a feeler connection.
    (https://en.wikipedia.org/wiki/Poisson_distribution)
    """

    def _poisson_next_send(self, now: float, avg_interval_seconds: int, random: Random) -> float:
        return now + (
            math.log(random.randrange(1 << 48) * -0.0000000000000035527136788 + 1) * avg_interval_seconds * -1000000.0
            + 0.5
        )

    async def _introducer_client(self) -> None:
        if self.introducer_info_obj is None:
            return None

        async def on_connect(peer: WSChiaConnection) -> None:
            msg = make_msg(ProtocolMessageTypes.request_peers_introducer, RequestPeersIntroducer())
            await peer.send_message(msg)

        await self.server.start_client(
            PeerInfo(await resolve(self.introducer_info_obj.host, prefer_ipv6=False), self.introducer_info_obj.port),
            on_connect,
        )

    async def _query_dns(self, dns_address: str) -> None:
        try:
            if self.default_port is None:
                self.log.error(
                    "Network id not supported in NETWORK_ID_DEFAULT_PORTS neither in config. Skipping DNS query."
                )
                return
            if self.resolver is None:
                self.log.warning("Skipping DNS query: asyncresolver not initialized.")
                return
            for rdtype in ["A", "AAAA"]:
                peers: list[TimestampedPeerInfo] = []
                result = await self.resolver.resolve(qname=dns_address, rdtype=rdtype, lifetime=30)
                for ip in result:
                    peers.append(
                        TimestampedPeerInfo(
                            ip.to_text(),
                            uint16(self.default_port),
                            uint64(0),
                        )
                    )
                self.log.info(f"Received {len(peers)} peers from DNS seeder, using rdtype = {rdtype}.")
                if len(peers) > 0:
                    await self._add_peers_common(peers, None, False)
        except Exception as e:
            self.log.warning(f"querying DNS introducer failed: {e}")

    async def on_connect_callback(self, peer: WSChiaConnection) -> None:
        if self.server.on_connect is not None:
            await self.server.on_connect(peer)
        else:
            await self.on_connect(peer)

    async def start_client_async(self, addr: PeerInfo, is_feeler: bool) -> None:
        try:
            if self.address_manager is None:
                return
            self.pending_outbound_connections.add(addr.host)
            client_connected = await self.server.start_client(
                addr,
                on_connect=self.on_connect_callback,
                is_feeler=is_feeler,
            )
            if self.server.is_duplicate_or_self_connection(addr):
                # Mark it as a softer attempt, without counting the failures.
                await self.address_manager.attempt(addr, False)
            elif client_connected is True:
                await self.address_manager.mark_good(addr)
                await self.address_manager.connect(addr)
            else:
                await self.address_manager.attempt(addr, True)
            self.pending_outbound_connections.remove(addr.host)
        except Exception as e:
            if addr.host in self.pending_outbound_connections:
                self.pending_outbound_connections.remove(addr.host)
            self.log.error(f"Exception in create outbound connections: {e}")
            self.log.error(f"Traceback: {traceback.format_exc()}")

    async def _connect_to_peers(self, random: Random) -> None:
        next_feeler = self._poisson_next_send(time.time() * 1000 * 1000, 240, random)
        retry_introducers = False
        dns_server_index: int = 0
        tried_all_dns_servers: bool = False
        local_peerinfo: PeerInfo | None = await self.server.get_peer_info()
        last_timestamp_local_info: uint64 = uint64(time.time())
        last_collision_timestamp = 0

        if self.initial_wait > 0:
            await asyncio.sleep(self.initial_wait)

        introducer_backoff = 1
        while not self.is_closed:
            try:
                assert self.address_manager is not None

                # We don't know any address, connect to the introducer to get some.
                size = await self.address_manager.size()
                if size == 0 or retry_introducers:
                    try:
                        await asyncio.sleep(introducer_backoff)
                    except asyncio.CancelledError:
                        return None
                    # Alternate between DNS servers and introducers.
                    # First try all the DNS servers in the list once. Then try the introducers once.
                    if len(self.dns_servers) > 0 and not tried_all_dns_servers:
                        dns_address = self.dns_servers[dns_server_index]
                        dns_server_index = (dns_server_index + 1) % len(self.dns_servers)
                        tried_all_dns_servers = dns_server_index == 0
                        await self._query_dns(dns_address)
                    else:
                        tried_all_dns_servers = False
                        await self._introducer_client()
                        # there's some delay between receiving the peers from the
                        # introducer until they get incorporated to prevent this
                        # loop for running one more time. Add this delay to ensure
                        # that once we get peers, we stop contacting the introducer.
                        try:
                            await asyncio.sleep(5)
                        except asyncio.CancelledError:
                            return None

                    retry_introducers = False
                    # keep doubling the introducer delay until we reach 5
                    # minutes
                    if introducer_backoff < 300:
                        introducer_backoff *= 2
                    continue
                else:
                    introducer_backoff = 1

                # Only connect out to one peer per network group (/16 for IPv4).
                groups = set()
                full_node_connected = self.server.get_connections(NodeType.FULL_NODE, outbound=True)
                connected = [c.get_peer_info() for c in full_node_connected]
                connected = [c for c in connected if c is not None]
                for conn in full_node_connected:
                    peer = conn.get_peer_info()
                    if peer is None:
                        continue
                    group = peer.get_group()
                    groups.add(group)

                # Feeler Connections
                #
                # Design goals:
                # * Increase the number of connectable addresses in the tried table.
                #
                # Method:
                # * Choose a random address from new and attempt to connect to it if we can connect
                # successfully it is added to tried.
                # * Start attempting feeler connections only after node finishes making outbound
                # connections.
                # * Only make a feeler connection once every few minutes.

                is_feeler = False
                has_collision = False
                if self._num_needed_peers() == 0:
                    if time.time() * 1000 * 1000 > next_feeler:
                        next_feeler = self._poisson_next_send(time.time() * 1000 * 1000, 240, random)
                        is_feeler = True

                await self.address_manager.resolve_tried_collisions()
                tries = 0
                now = time.time()
                got_peer = False
                addr: PeerInfo | None = None
                max_tries = 50
                if len(groups) < 3:
                    max_tries = 10
                elif len(groups) <= 5:
                    max_tries = 25
                select_peer_interval = max(0.1, len(groups) * 0.25)
                while not got_peer and not self.is_closed:
                    self.log.debug(f"Address manager query count: {tries}. Query limit: {max_tries}")
                    try:
                        await asyncio.sleep(select_peer_interval)
                    except asyncio.CancelledError:
                        return None
                    tries += 1
                    if tries > max_tries:
                        addr = None
                        retry_introducers = True
                        break
                    info: ExtendedPeerInfo | None = await self.address_manager.select_tried_collision()
                    if info is None or time.time() - last_collision_timestamp <= 60:
                        info = await self.address_manager.select_peer(is_feeler)
                    else:
                        has_collision = True
                        last_collision_timestamp = int(time.time())
                    if info is None:
                        if not is_feeler:
                            retry_introducers = True
                        break
                    # Require outbound connections, other than feelers,
                    # to be to distinct network groups.
                    addr = info.peer_info
                    if has_collision:
                        break
                    if not is_feeler and addr.get_group() in groups:
                        addr = None
                        continue
                    if addr in connected:
                        addr = None
                        continue
                    # attempt a node once per 30 minutes.
                    if now - info.last_try < 1800:
                        continue
                    if time.time() - last_timestamp_local_info > 1800 or local_peerinfo is None:
                        local_peerinfo = await self.server.get_peer_info()
                        last_timestamp_local_info = uint64(time.time())
                    if local_peerinfo is not None and addr == local_peerinfo:
                        continue
                    got_peer = True
                    self.log.debug(f"Addrman selected address: {addr}.")

                disconnect_after_handshake = is_feeler
                extra_peers_needed = self._num_needed_peers()
                if extra_peers_needed == 0:
                    disconnect_after_handshake = True
                    retry_introducers = False
                self.log.debug(f"Num peers needed: {extra_peers_needed}")
                initiate_connection = extra_peers_needed > 0 or has_collision or is_feeler
                connect_peer_interval = max(0.25, len(groups) * 0.5)
                if not initiate_connection:
                    connect_peer_interval += 15
                connect_peer_interval = min(connect_peer_interval, self.peer_connect_interval)
                if addr is not None and initiate_connection and addr.host not in self.pending_outbound_connections:
                    if len(self.pending_outbound_connections) >= MAX_CONCURRENT_OUTBOUND_CONNECTIONS:
                        self.log.debug("Max concurrent outbound connections reached. waiting")
                        await asyncio.wait(self.pending_tasks, return_when=asyncio.FIRST_COMPLETED)
                    self.pending_tasks.add(
                        create_referenced_task(self.start_client_async(addr, disconnect_after_handshake))
                    )

                await asyncio.sleep(connect_peer_interval)

                # prune completed connect tasks
                self.pending_tasks = set(filter(lambda t: not t.done(), self.pending_tasks))

            except Exception as e:
                self.log.error(f"Exception in create outbound connections: {e}")
                self.log.error(f"Traceback: {traceback.format_exc()}")

    async def _periodically_serialize(self, random: Random) -> None:
        while not self.is_closed:
            if self.address_manager is None:
                await asyncio.sleep(10)
                continue
            serialize_interval = random.randint(15 * 60, 30 * 60)
            await asyncio.sleep(serialize_interval)
            async with self.address_manager.lock:
                serialised_bytes = self.address_manager.serialize_bytes()
            await write_file_async(self.peers_file_path, serialised_bytes, file_mode=0o644)

    async def _periodically_cleanup(self) -> None:
        while not self.is_closed:
            # Removes entries with timestamp worse than 14 days ago
            # and with a high number of failed attempts.
            # Most likely, the peer left the network,
            # so we can save space in the peer tables.
            cleanup_interval = 1800
            max_timestamp_difference = 14 * 3600 * 24
            max_consecutive_failures = 10
            await asyncio.sleep(cleanup_interval)

            # Perform the cleanup only if we have at least 3 connections.
            full_node_connected = self.server.get_connections(NodeType.FULL_NODE)
            connected = [c.get_peer_info() for c in full_node_connected]
            connected = [c for c in connected if c is not None]
            if self.address_manager is not None and len(connected) >= 3:
                async with self.address_manager.lock:
                    self.address_manager.cleanup(max_timestamp_difference, max_consecutive_failures)

    def _peer_has_wrong_network_port(self, port: uint16) -> bool:
        # Check if the peer is having the default port of a network different than ours.
        return port in NETWORK_ID_DEFAULT_PORTS.values() and port != self.default_port

    async def _add_peers_common(
        self, peer_list: list[TimestampedPeerInfo], peer_src: PeerInfo | None, is_full_node: bool
    ) -> None:
        # Check if we got the peers from a full node or from the introducer.
        peers_adjusted_timestamp = []
        is_misbehaving = False
        if len(peer_list) > MAX_PEERS_RECEIVED_PER_REQUEST:
            is_misbehaving = True
        if is_full_node:
            if peer_src is None:
                return None
            async with self.lock:
                if peer_src.host not in self.received_count_from_peers:
                    self.received_count_from_peers[peer_src.host] = 0
                self.received_count_from_peers[peer_src.host] += len(peer_list)
                if self.received_count_from_peers[peer_src.host] > MAX_TOTAL_PEERS_RECEIVED:
                    is_misbehaving = True
        if is_misbehaving:
            return None
        for peer in peer_list:
            # Fast-fail oversized junk before IP parsing. IPAddress.create() rejects
            # invalid input too, but parsing pathological long strings is much slower.
            if len(peer.host) > MAX_IP_ADDRESS_STRING_LENGTH:
                self.log.debug(f"Skipping peer with oversized host string ({len(peer.host)} chars)")
                continue
            try:
                IPAddress.create(peer.host)
            except ValueError:
                self.log.debug(f"Skipping peer with invalid host: {peer.host!r:.50}")
                continue
            if peer.timestamp < 100000000 or peer.timestamp > time.time() + 10 * 60:
                # Invalid timestamp, predefine a bad one.
                current_peer = TimestampedPeerInfo(
                    peer.host,
                    peer.port,
                    uint64(time.time() - 5 * 24 * 60 * 60),
                )
            else:
                current_peer = peer
            if not is_full_node:
                current_peer = TimestampedPeerInfo(
                    peer.host,
                    peer.port,
                    uint64(0),
                )

            if not self._peer_has_wrong_network_port(peer.port):
                peers_adjusted_timestamp.append(current_peer)

        assert self.address_manager is not None

        if is_full_node:
            await self.address_manager.add_to_new_table(peers_adjusted_timestamp, peer_src, 2 * 60 * 60)
        else:
            await self.address_manager.add_to_new_table(peers_adjusted_timestamp, None, 0)


@dataclass
class FullNodePeers(FullNodeDiscovery):
    self_advertise_task: asyncio.Task[None] | None = field(default=None)
    address_relay_task: asyncio.Task[None] | None = field(default=None)
    relay_queue: asyncio.Queue[tuple[TimestampedPeerInfo, int]] = field(default_factory=asyncio.Queue)
    neighbour_known_peers: dict[PeerInfo, set[str]] = field(default_factory=dict)
    key: int = field(default_factory=lambda: random.getrandbits(256))

    def __post_init__(self) -> None:
        super().__post_init__()

    @contextlib.asynccontextmanager
    async def manage(self) -> AsyncIterator[None]:
        try:
            self.log.info("Initialising Full Node peers discovery.")
            await self.initialize_address_manager()
            self.self_advertise_task = create_referenced_task(self._periodically_self_advertise_and_clean_data())
            self.address_relay_task = create_referenced_task(self._address_relay())
            await self.start_tasks()
            self.log.info("Successfully initialised Full Node peers discovery.")
            yield
        finally:
            self.log.info("Closing Full Node peers discovery.")
            await self._close_common()
            cancel_task_safe(self.self_advertise_task, self.log)
            cancel_task_safe(self.address_relay_task, self.log)
            self.log.info("Successfully closed Full Node peers discovery.")

    async def _periodically_self_advertise_and_clean_data(self) -> None:
        while not self.is_closed:
            try:
                try:
                    await asyncio.sleep(24 * 3600)
                except asyncio.CancelledError:
                    return None
                # Clean up known nodes for neighbours every 24 hours.
                async with self.lock:
                    for neighbour, known_peers in self.neighbour_known_peers.items():
                        known_peers.clear()
                # Self advertise every 24 hours.
                peer = await self.server.get_peer_info()
                if peer is None:
                    continue
                timestamped_peer = [
                    TimestampedPeerInfo(
                        peer.host,
                        peer.port,
                        uint64(time.time()),
                    )
                ]
                msg = make_msg(
                    ProtocolMessageTypes.respond_peers,
                    RespondPeers(timestamped_peer),
                )
                await self.server.send_to_all([msg], NodeType.FULL_NODE)

                async with self.lock:
                    for host in list(self.received_count_from_peers.keys()):
                        self.received_count_from_peers[host] = 0
            except Exception as e:
                self.log.error(f"Exception in self advertise: {e}")
                self.log.error(f"Traceback: {traceback.format_exc()}")

    async def add_peers_neighbour(self, peers: list[TimestampedPeerInfo], neighbour_info: PeerInfo) -> None:
        async with self.lock:
            for peer in peers:
                if len(peer.host) > MAX_IP_ADDRESS_STRING_LENGTH:
                    continue
                try:
                    IPAddress.create(peer.host)
                except ValueError:
                    continue
                if neighbour_info not in self.neighbour_known_peers:
                    self.neighbour_known_peers[neighbour_info] = set()
                if peer.host not in self.neighbour_known_peers[neighbour_info]:
                    self.neighbour_known_peers[neighbour_info].add(peer.host)

    async def request_peers(self, peer_info: PeerInfo) -> Message | None:
        try:
            # Prevent a fingerprint attack: do not send peers to inbound connections.
            # This asymmetric behavior for inbound and outbound connections was introduced
            # to prevent a fingerprinting attack: an attacker can send specific fake addresses
            # to users' AddrMan and later request them by sending getaddr messages.
            # Making nodes which are behind NAT and can only make outgoing connections ignore
            # the request_peers message mitigates the attack.
            if self.address_manager is None:
                return None
            peers = await self.address_manager.get_peers()
            peers = [peer for peer in peers if not self._peer_has_wrong_network_port(peer.port)]
            await self.add_peers_neighbour(peers, peer_info)

            msg = make_msg(
                ProtocolMessageTypes.respond_peers,
                RespondPeers(peers),
            )

            return msg
        except Exception as e:
            self.log.error(f"Request peers exception: {e}")
            return None

    async def add_peers(
        self, peer_list: list[TimestampedPeerInfo], peer_src: PeerInfo | None, is_full_node: bool
    ) -> None:
        try:
            await self._add_peers_common(peer_list, peer_src, is_full_node)
            if is_full_node:
                if peer_src is None:
                    return
                await self.add_peers_neighbour(peer_list, peer_src)
                if len(peer_list) == 1 and self.relay_queue is not None:
                    peer = peer_list[0]
                    if peer.timestamp > time.time() - 60 * 10 and not self._peer_has_wrong_network_port(peer.port):
                        self.relay_queue.put_nowait((peer, 2))
        except Exception as e:
            self.log.error(f"Respond peers exception: {e}. Traceback: {traceback.format_exc()}")
        return None

    async def _address_relay(self) -> None:
        while not self.is_closed:
            try:
                try:
                    assert self.relay_queue is not None, "FullNodePeers.relay_queue should always exist"
                    relay_peer, num_peers = await self.relay_queue.get()
                except asyncio.CancelledError:
                    return None
                try:
                    IPAddress.create(relay_peer.host)
                except ValueError:
                    continue
                # https://en.bitcoin.it/wiki/Satoshi_Client_Node_Discovery#Address_Relay
                connections = self.server.get_connections(NodeType.FULL_NODE)
                hashes = []
                cur_day = int(time.time()) // (24 * 60 * 60)
                for connection in connections:
                    peer_info = connection.get_peer_info()
                    if peer_info is None:
                        continue
                    cur_hash = int.from_bytes(
                        bytes(
                            std_hash(
                                self.key.to_bytes(32, byteorder="big")
                                + peer_info.get_key()
                                + cur_day.to_bytes(3, byteorder="big")
                            )
                        ),
                        byteorder="big",
                    )
                    hashes.append((cur_hash, connection))
                hashes.sort(key=lambda x: x[0])
                for index, (_, connection) in enumerate(hashes):
                    if index >= num_peers:
                        break
                    peer_info = connection.get_peer_info()
                    if peer_info is None:
                        continue
                    async with self.lock:
                        if peer_info not in self.neighbour_known_peers:
                            self.neighbour_known_peers[peer_info] = set()
                        known_peers = self.neighbour_known_peers[peer_info]
                        if relay_peer.host in known_peers:
                            continue
                        known_peers.add(relay_peer.host)
                    if connection.peer_node_id is None:
                        continue
                    msg = make_msg(
                        ProtocolMessageTypes.respond_peers,
                        RespondPeers([relay_peer]),
                    )
                    await connection.send_message(msg)
            except Exception as e:
                self.log.error(f"Exception in address relay: {e}")
                self.log.error(f"Traceback: {traceback.format_exc()}")


@dataclass
class WalletPeers(FullNodeDiscovery):
    def __post_init__(self) -> None:
        super().__post_init__()

    async def start(self) -> None:
        self.initial_wait = 1
        await self.initialize_address_manager()
        await self.start_tasks()

    async def ensure_is_closed(self) -> None:
        if self.is_closed:
            return None
        await self._close_common()

    async def add_peers(
        self, peer_list: list[TimestampedPeerInfo], peer_src: PeerInfo | None, is_full_node: bool
    ) -> None:
        await self._add_peers_common(peer_list, peer_src, is_full_node)