From 613cdc47f2236199a17b4f2b9e0fd5e8071e414e Mon Sep 17 00:00:00 2001 From: Mariano Sorgente <3069354+mariano54@users.noreply.github.com> Date: Mon, 7 Feb 2022 19:38:45 -0500 Subject: [PATCH] Ms.wallet fixes (#10094) * wallet fixes * Don't show false positive synched * Code cleanup and lint * Fixes * Revert issue * Fix reorg issue Co-authored-by: wjblanke --- chia/cmds/configure.py | 1 + chia/wallet/wallet_node.py | 138 +++++++++++++++++----------- chia/wallet/wallet_state_manager.py | 27 +++--- 3 files changed, 96 insertions(+), 70 deletions(-) diff --git a/chia/cmds/configure.py b/chia/cmds/configure.py index 4741862232..4c28481956 100644 --- a/chia/cmds/configure.py +++ b/chia/cmds/configure.py @@ -105,6 +105,7 @@ def configure( config["introducer"]["port"] = int(testnet_port) config["full_node"]["introducer_peer"]["host"] = testnet_introducer config["full_node"]["dns_servers"] = [testnet_dns_introducer] + config["wallet"]["dns_servers"] = [testnet_dns_introducer] config["selected_network"] = testnet config["harvester"]["selected_network"] = testnet config["pool"]["selected_network"] = testnet diff --git a/chia/wallet/wallet_node.py b/chia/wallet/wallet_node.py index c0dc82d0cc..483458e5b0 100644 --- a/chia/wallet/wallet_node.py +++ b/chia/wallet/wallet_node.py @@ -472,37 +472,54 @@ class WalletNode: self.wallet_state_manager.state_changed("coin_added", wallet_id) self.synced_peers.add(full_node.peer_node_id) - async def receive_state_from_untrusted_peer(self, items: List[CoinState], peer, height: Optional[uint32]): + async def receive_state_from_peer( + self, items: List[CoinState], peer: WSChiaConnection, fork_height: Optional[uint32], height: Optional[uint32] + ): + assert self.wallet_state_manager is not None + trusted = self.is_trusted(peer) # Validate states in parallel, apply serial if self.validation_semaphore is None: self.validation_semaphore = asyncio.Semaphore(6) if self.new_state_lock is None: self.new_state_lock = asyncio.Lock() + # If there is a fork, we need to ensure that we roll back in trusted mode to properly handle reorgs + if trusted and fork_height is not None and height is not None and fork_height != height - 1: + await self.wallet_state_manager.reorg_rollback(fork_height) + all_tasks = [] - for idx, state in enumerate(items): + for idx, potential_state in enumerate(items): - async def receive_and_validate(state: CoinState, peer, height: Optional[uint32], idx): + async def receive_and_validate(inner_state: CoinState, inner_idx: int): assert self.wallet_state_manager is not None assert self.validation_semaphore is not None # if height is not None: async with self.validation_semaphore: - valid = await self.validate_received_state_from_peer(state, peer, self.get_cache_for_peer(peer)) - if valid: - self.log.info(f"new coin state received ({idx} / {len(items)})") - assert self.new_state_lock is not None - async with self.new_state_lock: - await self.wallet_state_manager.new_coin_state([state], peer) - elif height is not None: - self.add_state_to_race_cache(height, state) - else: - if state.created_height is not None: - self.add_state_to_race_cache(state.created_height, state) - if state.spent_height is not None: - self.add_state_to_race_cache(state.spent_height, state) + try: + if trusted: + valid = True + else: + valid = await self.validate_received_state_from_peer( + inner_state, peer, self.get_cache_for_peer(peer) + ) + if valid: + self.log.info(f"new coin state received ({inner_idx} / {len(items)})") + assert self.new_state_lock is not None + async with self.new_state_lock: + await self.wallet_state_manager.new_coin_state([inner_state], peer) + elif height is not None: + self.add_state_to_race_cache(height, inner_state) + else: + if inner_state.created_height is not None: + self.add_state_to_race_cache(inner_state.created_height, inner_state) + if inner_state.spent_height is not None: + self.add_state_to_race_cache(inner_state.spent_height, inner_state) + except Exception as e: + tb = traceback.format_exc() + self.log.error(f"Exception while adding state: {e} {tb}") - task = receive_and_validate(state, peer, height, idx) + task = receive_and_validate(potential_state, idx) all_tasks.append(task) while len(self.validation_semaphore._waiters) > 20: self.log.debug("sleeping 2 sec") @@ -518,25 +535,19 @@ class WalletNode: assert self.wallet_state_manager is not None msg = wallet_protocol.RegisterForPhUpdates(puzzle_hashes, height) all_state: Optional[RespondToPhUpdates] = await peer.register_interest_in_puzzle_hash(msg) - # State for untrusted sync is processed only in wp sync | or short sync backwards if all_state is None: return + await self.receive_state_from_peer(all_state.coin_states, peer, None, None) - if self.is_trusted(peer): - await self.wallet_state_manager.new_coin_state(all_state.coin_states, peer) - else: - await self.receive_state_from_untrusted_peer(all_state.coin_states, peer, None) - - async def subscribe_to_coin_updates(self, coin_names, peer, height=uint32(0)): + async def subscribe_to_coin_updates(self, coin_names: List[bytes32], peer: WSChiaConnection, height=uint32(0)): """ Tell full nodes that we are interested in coin ids, and for trusted connections, add the new coin state for the coin changes. """ msg = wallet_protocol.RegisterForCoinUpdates(coin_names, height) all_coins_state: Optional[RespondToCoinUpdates] = await peer.register_interest_in_coin(msg) - # State for untrusted sync is processed only in wp sync | or short sync backwards if all_coins_state is not None and self.is_trusted(peer): - await self.wallet_state_manager.new_coin_state(all_coins_state.coin_states, peer) + await self.receive_state_from_peer(all_coins_state.coin_states, peer, None, None) async def get_coin_state( self, coin_names: List[bytes32], peer: Optional[WSChiaConnection] = None @@ -594,19 +605,13 @@ class WalletNode: assert self.wallet_state_manager is not None assert self.server is not None - if self.is_trusted(peer): - async with self.new_peak_lock: - async with self.wallet_state_manager.lock: - self.log.debug(f"state_update_received is {request}") - await self.wallet_state_manager.new_coin_state( - request.items, peer, request.fork_height, request.height - ) - await self.update_ui() - else: - async with self.new_peak_lock: - async with self.wallet_state_manager.lock: - self.log.debug(f"state_update_received is {request}") - await self.receive_state_from_untrusted_peer(request.items, peer, request.height) + async with self.new_peak_lock: + async with self.wallet_state_manager.lock: + self.log.debug(f"state_update_received is {request}") + await self.receive_state_from_peer( + request.items, peer, request.fork_height if self.is_trusted(peer) else None, request.height + ) + await self.update_ui() def get_full_node_peer(self): nodes = self.server.get_full_node_connections() @@ -689,6 +694,7 @@ class WalletNode: curr_height = uint32(curr_height - 1) async def new_peak_wallet(self, peak: wallet_protocol.NewPeakWallet, peer: WSChiaConnection): + self.log.info(f"New peak wallet.. {peak.height} {peer.get_peer_info()}") assert self.wallet_state_manager is not None assert self.server is not None request_time = int(time.time()) @@ -846,11 +852,12 @@ class WalletNode: self.wallet_state_manager.set_sync_mode(False) else: + self.log.info(f"Starting backtrack sync to {peer.get_peer_info()}") await self.wallet_short_sync_backtrack(peak_block, peer) if peer.peer_node_id not in self.synced_peers: # Edge case, we still want to subscribe for all phs # (Hints are not in filter) - await self.untrusted_subscribe_to_puzzle_hashes(peer, True, self.get_cache_for_peer(peer)) + await self.untrusted_subscribe_to_puzzle_hashes(peer, self.get_cache_for_peer(peer)) self.synced_peers.add(peer.peer_node_id) if peak_block.height in self.race_cache: @@ -860,10 +867,17 @@ class WalletNode: ) if valid: await self.wallet_state_manager.new_coin_state([state], peer) + else: + self.log.warning(f"Invalid state from peer {peer}") + await peer.close(9999) + return self.wallet_state_manager.set_sync_mode(False) self.wallet_state_manager.state_changed("new_block") await self.wallet_state_manager.new_peak(peak) + + if peak.height > self.wallet_state_manager.finished_sync_up_to: + self.wallet_state_manager.finished_sync_up_to = uint32(peak.height) self._pending_tx_handler() async def wallet_short_sync_backtrack(self, header_block: HeaderBlock, peer) -> int: @@ -888,7 +902,13 @@ class WalletNode: fork_height = top.height - 1 blocks.reverse() - await self.wallet_state_manager.reorg_rollback(fork_height) + + # Roll back coins and transactions + peak_height = self.wallet_state_manager.blockchain.get_peak_height() + if fork_height < peak_height: + self.log.info(f"Rolling back to {fork_height}") + await self.wallet_state_manager.reorg_rollback(fork_height) + peak = await self.wallet_state_manager.blockchain.get_peak_block() self.rollback_request_caches(fork_height) @@ -957,7 +977,6 @@ class WalletNode: async def untrusted_subscribe_to_puzzle_hashes( self, peer: WSChiaConnection, - save_state: bool, peer_request_cache: Optional[PeerRequestCache], ): assert self.wallet_state_manager is not None @@ -978,9 +997,8 @@ class WalletNode: all_state: Optional[RespondToPhUpdates] = await peer.register_interest_in_puzzle_hash(msg) assert all_state is not None - if save_state: - assert peer_request_cache is not None - await self.receive_state_from_untrusted_peer(all_state.coin_states, peer, None) + assert peer_request_cache is not None + await self.receive_state_from_peer(all_state.coin_states, peer, None, None) # Check if new puzzle hashed have been created check_again = await self.get_puzzle_hashes_to_subscribe() @@ -1005,12 +1023,12 @@ class WalletNode: # Always sync fully from untrusted # Get state for puzzle hashes self.log.debug("Start untrusted_subscribe_to_puzzle_hashes ") - await self.untrusted_subscribe_to_puzzle_hashes(peer, True, peer_request_cache) + await self.untrusted_subscribe_to_puzzle_hashes(peer, peer_request_cache) self.log.debug("End untrusted_subscribe_to_puzzle_hashes ") - checked_call_coins = False + checked_all_coins = False checked_coins: Set[bytes32] = set() - while not checked_call_coins: + while not checked_all_coins: # Get state for coins ids all_coins = await self.wallet_state_manager.coin_store.get_coins_to_check(uint32(0)) all_coin_names = [coin_record.name() for coin_record in all_coins] @@ -1046,17 +1064,17 @@ class WalletNode: if coin_state_entry.created_height <= fork_height: coin_state_before_fork.append(coin_state_entry) - await self.receive_state_from_untrusted_peer(coin_state_before_fork, peer, None) + await self.receive_state_from_peer(coin_state_before_fork, peer, None, None) all_coins = await self.wallet_state_manager.coin_store.get_coins_to_check(uint32(0)) all_coin_names = [coin_record.name() for coin_record in all_coins] removed_dict = await self.wallet_state_manager.trade_manager.get_coins_of_interest() all_coin_names.extend(removed_dict.keys()) - checked_call_coins = True + checked_all_coins = True for coin_name in all_coin_names: if coin_name not in checked_coins: - checked_call_coins = False + checked_all_coins = False break end_time = time.time() @@ -1066,7 +1084,7 @@ class WalletNode: async def validate_received_state_from_peer( self, coin_state: CoinState, - peer, + peer: WSChiaConnection, peer_request_cache: PeerRequestCache, ) -> bool: """ @@ -1187,7 +1205,9 @@ class WalletNode: peer_request_cache.states_validated[coin_state.coin.get_hash()] = coin_state return True - async def validate_block_inclusion(self, block: HeaderBlock, peer, peer_request_cache: PeerRequestCache) -> bool: + async def validate_block_inclusion( + self, block: HeaderBlock, peer: WSChiaConnection, peer_request_cache: PeerRequestCache + ) -> bool: assert self.wallet_state_manager is not None if self.wallet_state_manager.blockchain.contains_height(block.height): stored_hash = self.wallet_state_manager.blockchain.height_to_hash(block.height) @@ -1242,22 +1262,28 @@ class WalletNode: self.log.error("Failed validation 2") return False - blocks = [] - + blocks: List[HeaderBlock] = [] for i in range(start - (start % 32), end + 1, 32): request_start = min(uint32(i), end) request_end = min(uint32(i + 31), end) request_h_response = RequestHeaderBlocks(request_start, request_end) if (request_start, request_end) in peer_request_cache.block_requests: - res_h_blocks: RespondHeaderBlocks = peer_request_cache.block_requests[(request_start, request_end)] + self.log.info(f"Using cache for blocks {request_start} - {request_end}") + res_h_blocks: Optional[RespondHeaderBlocks] = peer_request_cache.block_requests[ + (request_start, request_end) + ] else: start_time = time.time() res_h_blocks = await peer.request_header_blocks(request_h_response) + if res_h_blocks is None: + self.log.error("Failed validation 2.5") + return False end_time = time.time() peer_request_cache.block_requests[(request_start, request_end)] = res_h_blocks self.log.info( f"Fetched blocks: {request_start} - {request_end} | duration: {end_time - start_time}" ) + assert res_h_blocks is not None blocks.extend([bl for bl in res_h_blocks.header_blocks if bl.height >= start]) if compare_to_recent and weight_proof.recent_chain_data[0].header_hash != blocks[-1].header_hash: diff --git a/chia/wallet/wallet_state_manager.py b/chia/wallet/wallet_state_manager.py index 91cd6be970..d3a060c1d9 100644 --- a/chia/wallet/wallet_state_manager.py +++ b/chia/wallet/wallet_state_manager.py @@ -16,6 +16,7 @@ from chia.pools.pool_puzzles import SINGLETON_LAUNCHER_HASH, solution_to_pool_st from chia.pools.pool_wallet import PoolWallet from chia.protocols import wallet_protocol from chia.protocols.wallet_protocol import PuzzleSolutionResponse, RespondPuzzleSolution, CoinState +from chia.server.ws_connection import WSChiaConnection from chia.types.blockchain_format.coin import Coin from chia.types.blockchain_format.program import Program from chia.types.blockchain_format.sized_bytes import bytes32 @@ -105,6 +106,7 @@ class WalletStateManager: blockchain: WalletBlockchain coin_store: WalletCoinStore sync_store: WalletSyncStore + finished_sync_up_to: uint32 interested_store: WalletInterestedStore weight_proof_handler: WalletWeightProofHandler server: ChiaServer @@ -155,6 +157,7 @@ class WalletStateManager: self.wallet_node = wallet_node self.sync_mode = False + self.finished_sync_up_to = uint32(0) self.weight_proof_handler = WalletWeightProofHandler(self.constants) self.blockchain = await WalletBlockchain.create(self.basic_store, self.constants, self.weight_proof_handler) @@ -443,6 +446,9 @@ class WalletStateManager: if latest is None: return False + if latest.height - self.finished_sync_up_to > 2: + return False + latest_timestamp = self.blockchain.get_latest_timestamp() if latest_timestamp > int(time.time()) - 10 * 60: @@ -620,9 +626,7 @@ class WalletStateManager: async def new_coin_state( self, coin_states: List[CoinState], - peer, - fork_height: Optional[uint32] = None, - current_height: Optional[uint32] = None, + peer: WSChiaConnection, ): created_h_none = [] for coin_st in coin_states.copy(): @@ -636,12 +640,6 @@ class WalletStateManager: all_unconfirmed: List[TransactionRecord] = await self.tx_store.get_all_unconfirmed() trade_coin_removed: List[CoinState] = [] - if fork_height is not None and current_height is not None and fork_height != current_height - 1: - # This only applies to trusted mode - await self.reorg_rollback(fork_height) - - new_interested_coin_ids: List[bytes32] = [] - for coin_state_idx, coin_state in enumerate(coin_states): info = await self.get_wallet_id_for_puzzle_hash(coin_state.coin.puzzle_hash) local_record: Optional[WalletCoinRecord] = await self.coin_store.get_coin_record(coin_state.coin.name()) @@ -840,7 +838,8 @@ class WalletStateManager: uint32(record.wallet_id), record.wallet_type, ) - new_interested_coin_ids.append(new_singleton_coin.name()) + await self.coin_store.set_spent(curr_coin_state.coin.name(), curr_coin_state.spent_height) + await self.interested_store.add_interested_coin_id(new_singleton_coin.name(), True) new_coin_state: List[CoinState] = await self.wallet_node.get_coin_state( [new_singleton_coin.name()] ) @@ -888,12 +887,10 @@ class WalletStateManager: await self.coin_added( coin_added, coin_state.spent_height, [], pool_wallet.id(), WalletType(pool_wallet.type()) ) - new_interested_coin_ids.append(coin_added.name()) + await self.interested_store.add_interested_coin_id(coin_added.name(), True) else: raise RuntimeError("All cases already handled") # Logic error, all cases handled - for new_coin_id in new_interested_coin_ids: - await self.add_interested_coin_id(new_coin_id) for coin_state_removed in trade_coin_removed: await self.trade_manager.coins_of_interest_farmed(coin_state_removed) @@ -927,7 +924,7 @@ class WalletStateManager: return True return False - async def get_wallet_id_for_puzzle_hash(self, puzzle_hash) -> Optional[Tuple[uint32, WalletType]]: + async def get_wallet_id_for_puzzle_hash(self, puzzle_hash: bytes32) -> Optional[Tuple[uint32, WalletType]]: info = await self.puzzle_store.wallet_info_for_puzzle_hash(puzzle_hash) if info is not None: wallet_id, wallet_type = info @@ -936,6 +933,8 @@ class WalletStateManager: interested_wallet_id = await self.interested_store.get_interested_puzzle_hash_wallet_id(puzzle_hash=puzzle_hash) if interested_wallet_id is not None: wallet_id = uint32(interested_wallet_id) + if wallet_id not in self.wallets.keys(): + self.log.warning(f"Do not have wallet {wallet_id} for puzzle_hash {puzzle_hash}") wallet_type = WalletType(self.wallets[uint32(wallet_id)].type()) return wallet_id, wallet_type return None