diff --git a/chia/_tests/wallet/test_wallet_node.py b/chia/_tests/wallet/test_wallet_node.py index 51169f2e1b..53f92310f3 100644 --- a/chia/_tests/wallet/test_wallet_node.py +++ b/chia/_tests/wallet/test_wallet_node.py @@ -2183,3 +2183,57 @@ async def test_collect_valid_states( ) assert [cs.coin.name() for cs in valid_states] == [good_coin_state.coin.name()] assert f"Failed to validate coin_state {bad_coin_state}" in caplog.text + + +@pytest.mark.anyio +@pytest.mark.parametrize( + "subscription_kind", + ["puzzle_hash", "coin_id"], +) +async def test_process_new_subscriptions_continues_after_peer_error( + setup_two_nodes_and_wallet: OldSimulatorsAndWallets, + self_hostname: str, + subscription_kind: str, +) -> None: + """One failing peer must not skip subscription registration on later peers.""" + [bad_fn_api, good_fn_api], [(wallet_node, wallet_server)], _ = setup_two_nodes_and_wallet + # Keep both peers; do not prefer a trusted peer that would disconnect the other. + wallet_node.config["trusted_peers"] = {} + + assert await wallet_server.start_client(PeerInfo(self_hostname, bad_fn_api.server.get_port()), None) + assert await wallet_server.start_client(PeerInfo(self_hostname, good_fn_api.server.get_port()), None) + await time_out_assert(10, lambda: len(wallet_server.get_connections(NodeType.FULL_NODE)) == 2) + + target = bytes32.random() + + async def register_for_ph_updates( + self: object, request: wallet_protocol.RegisterForPhUpdates, peer: WSChiaConnection + ) -> None: + raise RuntimeError("simulated peer failure") + + async def register_for_coin_updates( + self: object, request: wallet_protocol.RegisterForCoinUpdates, peer: WSChiaConnection + ) -> None: + raise RuntimeError("simulated peer failure") + + if subscription_kind == "puzzle_hash": + handler: Any = register_for_ph_updates + + def good_peer_subscribed() -> bool: + return target in good_fn_api.full_node.subscriptions.puzzle_subscriptions(wallet_server.node_id) + + else: + handler = register_for_coin_updates + + def good_peer_subscribed() -> bool: + return target in good_fn_api.full_node.subscriptions.coin_subscriptions(wallet_server.node_id) + + with patch_request_handler(api=bad_fn_api, handler=handler): + if subscription_kind == "puzzle_hash": + await wallet_node.new_peak_queue.subscribe_to_puzzle_hashes([target]) + else: + await wallet_node.new_peak_queue.subscribe_to_coin_ids([target]) + await time_out_assert(10, good_peer_subscribed) + await time_out_assert(10, lambda: bad_fn_api.server.node_id not in wallet_server.all_connections) + + assert good_fn_api.server.node_id in wallet_server.all_connections diff --git a/chia/_tests/wallet/test_wallet_sync_utils.py b/chia/_tests/wallet/test_wallet_sync_utils.py new file mode 100644 index 0000000000..a66bc69926 --- /dev/null +++ b/chia/_tests/wallet/test_wallet_sync_utils.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +import pytest +from chia_rs.sized_ints import int16 + +from chia.protocols.shared_protocol import Error +from chia.wallet.util.wallet_sync_utils import _coin_states_from_subscribe_response + + +@pytest.mark.parametrize( + "response, match", + [ + (None, "None response from peer peer.example for register_for_ph_updates"), + ( + Error(int16(1), "rejected", None), + "Error response from peer peer.example for register_for_ph_updates", + ), + (object(), "Unexpected response from peer peer.example for register_for_ph_updates: object"), + ], + ids=["none", "protocol_error", "unexpected_type"], +) +def test_coin_states_from_subscribe_response_rejects_invalid(response: object, match: str) -> None: + with pytest.raises(ValueError, match=match): + _coin_states_from_subscribe_response(response, "peer.example", "register_for_ph_updates") diff --git a/chia/wallet/util/wallet_sync_utils.py b/chia/wallet/util/wallet_sync_utils.py index a56cbf4b74..1e3e6c7ea3 100644 --- a/chia/wallet/util/wallet_sync_utils.py +++ b/chia/wallet/util/wallet_sync_utils.py @@ -18,7 +18,7 @@ from chia_rs.sized_bytes import bytes32 from chia_rs.sized_ints import uint32 from chia.full_node.full_node_api import FullNodeAPI -from chia.protocols.shared_protocol import Capability +from chia.protocols.shared_protocol import Capability, Error from chia.protocols.wallet_protocol import ( RegisterForCoinUpdates, RegisterForPhUpdates, @@ -51,6 +51,21 @@ class PeerRequestException(Exception): pass +def _coin_states_from_subscribe_response( + response: object, + peer_host: str, + request_name: str, +) -> list[CoinState]: + """Normalize call_api results for subscription helpers into coin states or ValueError.""" + if response is None: + raise ValueError(f"None response from peer {peer_host} for {request_name}") + if isinstance(response, Error): + raise ValueError(f"Error response from peer {peer_host} for {request_name}: {response}") + if isinstance(response, (RespondToPhUpdates, RespondToCoinUpdates)): + return response.coin_states + raise ValueError(f"Unexpected response from peer {peer_host} for {request_name}: {type(response).__name__}") + + async def subscribe_to_phs( puzzle_hashes: list[bytes32], peer: WSChiaConnection, @@ -61,12 +76,10 @@ async def subscribe_to_phs( Tells full nodes that we are interested in puzzle hashes, and returns the response. """ msg = RegisterForPhUpdates(puzzle_hashes, uint32(max(min_height, uint32(0)))) - all_coins_state: RespondToPhUpdates | None = await peer.call_api( + all_coins_state: RespondToPhUpdates | Error | None = await peer.call_api( FullNodeAPI.register_for_ph_updates, msg, timeout=300, priority=priority ) - if all_coins_state is None: - raise ValueError(f"None response from peer {peer.peer_info.host} for register_for_ph_updates") - return all_coins_state.coin_states + return _coin_states_from_subscribe_response(all_coins_state, peer.peer_info.host, "register_for_ph_updates") async def subscribe_to_coin_updates( @@ -79,13 +92,10 @@ async def subscribe_to_coin_updates( Tells full nodes that we are interested in coin ids, and returns the response. """ msg = RegisterForCoinUpdates(coin_names, uint32(max(0, min_height))) - all_coins_state: RespondToCoinUpdates | None = await peer.call_api( + all_coins_state: RespondToCoinUpdates | Error | None = await peer.call_api( FullNodeAPI.register_for_coin_updates, msg, timeout=300, priority=priority ) - - if all_coins_state is None: - raise ValueError(f"None response from peer {peer.peer_info.host} for register_for_coin_updates") - return all_coins_state.coin_states + return _coin_states_from_subscribe_response(all_coins_state, peer.peer_info.host, "register_for_coin_updates") def validate_additions( diff --git a/chia/wallet/wallet_node.py b/chia/wallet/wallet_node.py index 13e983b523..bca0be7984 100644 --- a/chia/wallet/wallet_node.py +++ b/chia/wallet/wallet_node.py @@ -49,12 +49,20 @@ from chia.types.weight_proof import WeightProof from chia.util.batches import to_batches from chia.util.config import lock_and_load_config, process_config_start_method, save_config from chia.util.db_wrapper import SQLITE_MAX_VARIABLE_NUMBER, manage_connection -from chia.util.errors import Err, KeychainIsEmpty, KeychainIsLocked, KeychainKeyNotFound, KeychainProxyConnectionFailure +from chia.util.errors import ( + Err, + KeychainIsEmpty, + KeychainIsLocked, + KeychainKeyNotFound, + KeychainProxyConnectionFailure, + ProtocolError, +) from chia.util.hash import std_hash from chia.util.keychain import Keychain +from chia.util.log_exceptions import log_exceptions from chia.util.path import path_from_root from chia.util.profiler import mem_profile_task, profile_task -from chia.util.streamable import Streamable, streamable +from chia.util.streamable import Streamable, StreamableError, streamable from chia.util.task_referencer import create_referenced_task from chia.wallet.puzzles.clawback.metadata import AutoClaimSettings from chia.wallet.transaction_record import TransactionRecord @@ -682,7 +690,26 @@ class WalletNode: # we might not be able to process some state. coin_ids: list[bytes32] = item.data for peer in self.server.get_connections(NodeType.FULL_NODE): - coin_states: list[CoinState] = await subscribe_to_coin_updates(coin_ids, peer, 0) + try: + # Peer/RPC failures only. subscribe_to_* raises ValueError on None/Error + # responses; call_api may raise ProtocolError/StreamableError; transport + # may raise OSError. Local apply errors should surface to the outer handler. + coin_states: list[CoinState] = await subscribe_to_coin_updates(coin_ids, peer, 0) + except (ValueError, ProtocolError, OSError, StreamableError) as e: + # Keep going so one bad peer cannot drop this batch for everyone else. + self.log.warning( + "COIN_ID_SUBSCRIPTION failed for peer %s: %s", + peer.peer_info.host, + e, + ) + with log_exceptions( + self.log, + consume=True, + message=f"Failed closing peer {peer.peer_info.host} after COIN_ID_SUBSCRIPTION error", + level=logging.WARNING, + ): + await peer.close(9999) + continue if len(coin_states) > 0: async with self.wallet_state_manager.lock: await self.add_states_from_peer(coin_states, peer) @@ -690,8 +717,28 @@ class WalletNode: self.log.debug("Pulled from queue: %s %s", item.item_type.name, item.data) puzzle_hashes: list[bytes32] = item.data for peer in self.server.get_connections(NodeType.FULL_NODE): - # Puzzle hash subscription - coin_states = await subscribe_to_phs(puzzle_hashes, peer, 0) + try: + # Peer/RPC failures only. subscribe_to_* raises ValueError on None/Error + # responses; call_api may raise ProtocolError/StreamableError; transport + # may raise OSError. Local apply errors should surface to the outer handler. + coin_states = await subscribe_to_phs(puzzle_hashes, peer, 0) + except (ValueError, ProtocolError, OSError, StreamableError) as e: + # Keep going so one bad peer cannot drop this batch for everyone else. + self.log.warning( + "PUZZLE_HASH_SUBSCRIPTION failed for peer %s: %s", + peer.peer_info.host, + e, + ) + with log_exceptions( + self.log, + consume=True, + message=( + f"Failed closing peer {peer.peer_info.host} after PUZZLE_HASH_SUBSCRIPTION error" + ), + level=logging.WARNING, + ): + await peer.close(9999) + continue if len(coin_states) > 0: async with self.wallet_state_manager.lock: await self.add_states_from_peer(coin_states, peer)