chia-blockchain/chia/wallet/wallet_node_api.py

from __future__ import annotations

import logging
from typing import TYPE_CHECKING, ClassVar

from chia_rs import RespondToPhUpdates

from chia.protocols import full_node_protocol, introducer_protocol, wallet_protocol
from chia.protocols.outbound_message import NodeType
from chia.server.api_protocol import ApiMetadata
from chia.server.ws_connection import WSChiaConnection
from chia.types.mempool_inclusion_status import MempoolInclusionStatus
from chia.util.errors import Err
from chia.wallet.wallet_node import WalletNode


class WalletNodeAPI:
    if TYPE_CHECKING:
        from chia.apis.wallet_stub import WalletNodeApiStub

        # Verify this class implements the WalletNodeApiStub protocol
        def _protocol_check(self: WalletNodeAPI) -> WalletNodeApiStub:
            return self

    log: logging.Logger
    wallet_node: WalletNode
    metadata: ClassVar[ApiMetadata] = ApiMetadata()

    def __init__(self, wallet_node: WalletNode) -> None:
        self.log = logging.getLogger(__name__)
        self.wallet_node = wallet_node

    def ready(self) -> bool:
        return self.wallet_node.logged_in

    @metadata.request()
    async def respond_removals(self, response: wallet_protocol.RespondRemovals) -> None:
        pass

    @metadata.request()
    async def reject_removals_request(self, response: wallet_protocol.RejectRemovalsRequest) -> None:
        """
        The full node has rejected our request for removals.
        """

    @metadata.request()
    async def reject_additions_request(self, response: wallet_protocol.RejectAdditionsRequest) -> None:
        """
        The full node has rejected our request for additions.
        """

    @metadata.request(peer_required=True, execute_task=True)
    async def new_peak_wallet(self, peak: wallet_protocol.NewPeakWallet, peer: WSChiaConnection) -> None:
        """
        The full node sent as a new peak
        """
        # For trusted peers check if there are untrusted peers, if so make sure to disconnect them if the trusted node
        # is synced.
        if self.wallet_node.is_trusted(peer):
            full_node_connections = self.wallet_node.server.get_connections(NodeType.FULL_NODE)
            untrusted_peers = [
                peer for peer in full_node_connections if not self.wallet_node.is_trusted(peer) and not peer.closed
            ]

            # Check for untrusted peers to avoid fetching the timestamp if not required
            if len(untrusted_peers) > 0:
                timestamp = await self.wallet_node.get_timestamp_for_height_from_peer(peak.height, peer)
            else:
                timestamp = None
            if timestamp is not None and self.wallet_node.is_timestamp_in_sync(timestamp):
                self.log.info("Connected to a synced trusted peer, disconnecting from all untrusted nodes.")
                # Stop peer discovery/connect tasks first
                if self.wallet_node.wallet_peers is not None:
                    await self.wallet_node.wallet_peers.ensure_is_closed()
                    self.wallet_node.wallet_peers = None
                # Then disconnect from all untrusted nodes
                for untrusted_peer in untrusted_peers:
                    await untrusted_peer.close()

        await self.wallet_node.new_peak_queue.new_peak_wallet(peak, peer)

    @metadata.request()
    async def reject_header_request(self, response: wallet_protocol.RejectHeaderRequest) -> None:
        """
        The full node has rejected our request for a header.
        """

    @metadata.request()
    async def respond_block_header(self, response: wallet_protocol.RespondBlockHeader) -> None:
        pass

    @metadata.request()
    async def respond_additions(self, response: wallet_protocol.RespondAdditions) -> None:
        pass

    @metadata.request()
    async def respond_proof_of_weight(self, response: full_node_protocol.RespondProofOfWeight) -> None:
        pass

    @metadata.request(peer_required=True)
    async def transaction_ack(self, ack: wallet_protocol.TransactionAck, peer: WSChiaConnection) -> None:
        """
        This is an ack for our previous SendTransaction call. This removes the transaction from
        the send queue if we have sent it to enough nodes.
        """
        async with self.wallet_node.wallet_state_manager.lock:
            assert peer.peer_node_id is not None
            name = peer.peer_node_id.hex()
            matched_in_flight_send = False
            if peer.peer_node_id in self.wallet_node._tx_messages_in_progress:
                in_progress = self.wallet_node._tx_messages_in_progress[peer.peer_node_id]
                if ack.txid in in_progress:
                    matched_in_flight_send = True
                    in_progress.remove(ack.txid)
                if in_progress == []:
                    del self.wallet_node._tx_messages_in_progress[peer.peer_node_id]

            # Ignore stale/unsolicited acknowledgements that don't correspond to an in-flight send.
            if not matched_in_flight_send:
                self.wallet_node.log.debug(
                    "Ignoring unsolicited transaction ack from peer %s for tx %s", name, ack.txid
                )
                return

            status = MempoolInclusionStatus(ack.status)
            try:
                wallet_state_manager = self.wallet_node.wallet_state_manager
            except RuntimeError as e:
                if "not assigned" in str(e):
                    return
                raise

            if status == MempoolInclusionStatus.SUCCESS:
                self.wallet_node.log.info(
                    f"SpendBundle has been received and accepted to mempool by the FullNode. {ack}"
                )
            elif status == MempoolInclusionStatus.PENDING:
                self.wallet_node.log.info(f"SpendBundle has been received (and is pending) by the FullNode. {ack}")
            else:
                if not self.wallet_node.is_trusted(peer) and ack.error == Err.NO_TRANSACTIONS_WHILE_SYNCING.name:
                    self.wallet_node.log.info(f"Peer {peer.get_peer_info()} is not synced, closing connection")
                    await peer.close()
                    return
                self.wallet_node.log.warning(f"SpendBundle has been rejected by the FullNode. {ack}")
            if ack.error is not None:
                await wallet_state_manager.remove_from_queue(ack.txid, name, status, Err[ack.error])
            else:
                await wallet_state_manager.remove_from_queue(ack.txid, name, status, None)

    @metadata.request(peer_required=True)
    async def respond_peers_introducer(
        self, request: introducer_protocol.RespondPeersIntroducer, peer: WSChiaConnection
    ) -> None:
        if self.wallet_node.wallet_peers is not None:
            await self.wallet_node.wallet_peers.add_peers(request.peer_list, peer.get_peer_info(), False)

        if peer.connection_type is NodeType.INTRODUCER:
            await peer.close()

    @metadata.request(peer_required=True)
    async def respond_peers(self, request: full_node_protocol.RespondPeers, peer: WSChiaConnection) -> None:
        if self.wallet_node.wallet_peers is None:
            return

        self.log.info(f"Wallet received {len(request.peer_list)} peers.")
        await self.wallet_node.wallet_peers.add_peers(request.peer_list, peer.get_peer_info(), True)

    @metadata.request()
    async def respond_puzzle_solution(self, request: wallet_protocol.RespondPuzzleSolution) -> None:
        self.log.error("Unexpected message `respond_puzzle_solution`. Peer might be slow to respond")

    @metadata.request()
    async def reject_puzzle_solution(self, request: wallet_protocol.RejectPuzzleSolution) -> None:
        self.log.warning(f"Reject puzzle solution: {request}")

    @metadata.request()
    async def respond_header_blocks(self, request: wallet_protocol.RespondHeaderBlocks) -> None:
        pass

    @metadata.request()
    async def respond_block_headers(self, request: wallet_protocol.RespondBlockHeaders) -> None:
        pass

    @metadata.request()
    async def reject_header_blocks(self, request: wallet_protocol.RejectHeaderBlocks) -> None:
        self.log.warning(f"Reject header blocks: {request}")

    @metadata.request()
    async def reject_block_headers(self, request: wallet_protocol.RejectBlockHeaders) -> None:
        pass

    @metadata.request(
        peer_required=True,
        execute_task=True,
        # Cap deserialization at max+1 so oversized updates are still detectable and banned in
        # state_update_received, without materializing an unbounded CoinState list.
        list_limits=lambda self, peer: {"items": self.wallet_node.max_coin_state_update_items() + 1},
    )
    async def coin_state_update(self, request: wallet_protocol.CoinStateUpdate, peer: WSChiaConnection) -> None:
        await self.wallet_node.new_peak_queue.full_node_state_updated(request, peer)

    # TODO: Review this hinting issue around this rust type not being a Streamable
    #       subclass, as you might expect it wouldn't be.  Maybe we can get the
    #       protocol working right back at the api.request definition.
    @metadata.request()  # type: ignore[type-var]
    async def respond_to_ph_updates(self, request: RespondToPhUpdates) -> None:
        pass

    @metadata.request()
    async def respond_to_coin_updates(self, request: wallet_protocol.RespondToCoinUpdates) -> None:
        pass

    @metadata.request()
    async def respond_children(self, request: wallet_protocol.RespondChildren) -> None:
        pass

    @metadata.request()
    async def respond_ses_hashes(self, request: wallet_protocol.RespondSESInfo) -> None:
        pass

    @metadata.request()
    async def respond_fee_estimates(self, request: wallet_protocol.RespondFeeEstimates) -> None:
        pass

    @metadata.request()
    async def respond_blocks(self, request: full_node_protocol.RespondBlocks) -> None:
        pass