CHIA-4264: Continue Wallet Subscriptions After a Single Peer Fails (#21245)

* Continue wallet subscriptions after a single peer fails

One bad peer in _process_new_subscriptions previously aborted the whole
batch, leaving later peers without the new PH/coin-id registrations.

Co-authored-by: Cursor <cursoragent@cursor.com>

* Use real peers in subscription peer-error test

Replace MagicMock peers with two full nodes and only fail the bad peer's register handlers.

Co-authored-by: Cursor <cursoragent@cursor.com>

* Narrow subscription peer-error handling to RPC failures

Only catch subscribe failures so local apply errors do not ban every peer, and guard peer.close so close exceptions cannot abort the batch.

Co-authored-by: Cursor <cursoragent@cursor.com>

* Tighten subscription failure handling to peer/RPC errors

Catch ValueError/ProtocolError/OSError instead of Exception, and use log_exceptions when closing a failed peer.

Co-authored-by: Cursor <cursoragent@cursor.com>

* Treat protocol Error and StreamableError as subscription peer failures

Normalize Error/unexpected subscribe responses to ValueError, and catch
StreamableError in the per-peer handler so corrupt payloads do not abort the batch.

Co-authored-by: Cursor <cursoragent@cursor.com>

* Add unit tests for subscribe response normalization helper.

Cover Error, unexpected-type, None, and success paths without mocking peers.

Co-authored-by: Cursor <cursoragent@cursor.com>

* Drop happy-path unit test for subscribe response helper.

Rejection cases are enough for the coverage gap; success is already exercised elsewhere.

Co-authored-by: Cursor <cursoragent@cursor.com>

---------

Co-authored-by: Cursor <cursoragent@cursor.com>

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

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")

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(

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)