Repository: Chia-Network/chia-blockchain
Pull request: Ms.wallet refactor 
This is a significant change in how syncing works in the wallet.  It is now organized as a queue where one of four things can get processed at once:
* New puzzle hash subscription
* New coin ID subscription
* new_peak_wallet message
* state_update message

The subscriptions are prioritized to ensure we don't miss data. State updates are cached by header hash (not height) and reorgs are now processed correctly.  Many different bugs and race conditions are fixed by this change. Also, there is a performance improvement in untrusted sync. 

A full node bug in coin store is also fixed (affected wallet protocol), where the min height was not being looked up correctly.

TODO:
- [x] fix flaky tests
- [x] Test on larger wallets, and test switching between trusted/untrusted
- [ ] Fix oom issue (also present in main)
- [ ] Potentially remove unnecessary fork_point argument

Problem statement:
[enhancement] improve and optimize light wallet sync performance
Improve the reliability of the light wallet sync that enable faster sync times and better handling of wallets with large number of coins (~ up to 500) and medium # of transactions (~ up to 2000) and ensure accuracy of what is reported in the wallet.

Values are arbitrary and only used to convey approximate ranges.

Hints from issue discussion:

Commit message: Merge branch 'main' into ms.wallet_refactor

# Conflicts:
#	chia/wallet/wallet_node.py
#	chia/wallet/wallet_state_manager.py

Patch to review:
diff --git a/chia/cmds/show.py b/chia/cmds/show.py
index c28197292411..20e47d33959a 100644
--- a/chia/cmds/show.py
+++ b/chia/cmds/show.py
@@ -1,8 +1,58 @@
-from typing import Any, Optional, Union
+from typing import Any, Optional, Union, Dict
 
 from chia.types.blockchain_format.sized_bytes import bytes32
 import click
 
+from chia.util.network import is_trusted_inner
+
+
+async def print_connections(client, time, NodeType, trusted_peers: Dict):
+    connections = await client.get_connections()
+    print("Connections:")
+    print("Type      IP                                     Ports       NodeID      Last Connect" + "      MiB Up|Dwn")
+    for con in connections:
+        last_connect_tuple = time.struct_time(time.localtime(con["last_message_time"]))
+        last_connect = time.strftime("%b %d %T", last_connect_tuple)
+        mb_down = con["bytes_read"] / (1024 * 1024)
+        mb_up = con["bytes_written"] / (1024 * 1024)
+
+        host = con["peer_host"]
+        # Strip IPv6 brackets
+        host = host.strip("[]")
+
+        trusted: bool = is_trusted_inner(host, con["node_id"], trusted_peers, False)
+        # Nodetype length is 9 because INTRODUCER will be deprecated
+        if NodeType(con["type"]) is NodeType.FULL_NODE:
+            peak_height = con.get("peak_height", None)
+            connection_peak_hash = con.get("peak_hash", None)
+            if connection_peak_hash is None:
+                connection_peak_hash = "No Info"
+            else:
+                if connection_peak_hash.startswith(("0x", "0X")):
+                    connection_peak_hash = connection_peak_hash[2:]
+                connection_peak_hash = f"{connection_peak_hash[:8]}..."
+            con_str = (
+                f"{NodeType(con['type']).name:9} {host:38} "
+                f"{con['peer_port']:5}/{con['peer_server_port']:<5}"
+                f" {con['node_id'].hex()[:8]}... "
+                f"{last_connect}  "
+                f"{mb_up:7.1f}|{mb_down:<7.1f}"
+                f"\n                                                 "
+            )
+            if peak_height is not None:
+                con_str += f"-Height: {peak_height:8.0f}    -Hash: {connection_peak_hash}    -Trusted: {trusted}"
+            else:
+                con_str += f"-Height: No Info    -Hash: {connection_peak_hash}     -Trusted: {trusted}"
+        else:
+            con_str = (
+                f"{NodeType(con['type']).name:9} {host:38} "
+                f"{con['peer_port']:5}/{con['peer_server_port']:<5}"
+                f" {con['node_id'].hex()[:8]}... "
+                f"{last_connect}  "
+                f"{mb_up:7.1f}|{mb_down:<7.1f}"
+            )
+        print(con_str)
+
 
 async def show_async(
     rpc_port: Optional[int],
@@ -15,10 +65,8 @@ async def show_async(
     block_by_header_hash: str,
 ) -> None:
     import aiohttp
-    import time
     import traceback
-
-    from time import localtime, struct_time
+    import time
     from typing import List, Optional
     from chia.consensus.block_record import BlockRecord
     from chia.rpc.full_node_rpc_client import FullNodeRpcClient
@@ -84,7 +132,7 @@ async def show_async(
                     while curr is not None and not curr.is_transaction_block:
                         curr = await client.get_block_record(curr.prev_hash)
                     peak_time = curr.timestamp
-                peak_time_struct = struct_time(localtime(peak_time))
+                peak_time_struct = time.struct_time(time.localtime(peak_time))
 
                 print(
                     "      Time:",
@@ -115,51 +163,8 @@ async def show_async(
             if show_connections:
                 print("")
         if show_connections:
-            connections = await client.get_connections()
-            print("Connections:")
-            print(
-                "Type      IP                                     Ports       NodeID      Last Connect"
-                + "      MiB Up|Dwn"
-            )
-            for con in connections:
-                last_connect_tuple = struct_time(localtime(con["last_message_time"]))
-                last_connect = time.strftime("%b %d %T", last_connect_tuple)
-                mb_down = con["bytes_read"] / (1024 * 1024)
-                mb_up = con["bytes_written"] / (1024 * 1024)
-
-                host = con["peer_host"]
-                # Strip IPv6 brackets
-                host = host.strip("[]")
-                # Nodetype length is 9 because INTRODUCER will be deprecated
-                if NodeType(con["type"]) is NodeType.FULL_NODE:
-                    peak_height = con["peak_height"]
-                    connection_peak_hash = con["peak_hash"]
-                    if connection_peak_hash is None:
-                        connection_peak_hash = "No Info"
-                    else:
-                        if connection_peak_hash.startswith(("0x", "0X")):
-                            connection_peak_hash = connection_peak_hash[2:]
-                        connection_peak_hash = f"{connection_peak_hash[:8]}..."
-                    if peak_height is None:
-                        peak_height = 0
-                    con_str = (
-                        f"{NodeType(con['type']).name:9} {host:38} "
-                        f"{con['peer_port']:5}/{con['peer_server_port']:<5}"
-                        f" {con['node_id'].hex()[:8]}... "
-                        f"{last_connect}  "
-                        f"{mb_up:7.1f}|{mb_down:<7.1f}"
-                        f"\n                                                 "
-                        f"-SB Height: {peak_height:8.0f}    -Hash: {connection_peak_hash}"
-                    )
-                else:
-                    con_str = (
-                        f"{NodeType(con['type']).name:9} {host:38} "
-                        f"{con['peer_port']:5}/{con['peer_server_port']:<5}"
-                        f" {con['node_id'].hex()[:8]}... "
-                        f"{last_connect}  "
-                        f"{mb_up:7.1f}|{mb_down:<7.1f}"
-                    )
-                print(con_str)
+            trusted_peers: Dict = config["full_node"].get("trusted_peers", {})
+            await print_connections(client, time, NodeType, trusted_peers)
             # if called together with state, leave a blank line
             if state:
                 print("")
@@ -217,8 +222,8 @@ async def show_async(
                     difficulty = block.weight
                 if block.is_transaction_block:
                     assert full_block.transactions_info is not None
-                    block_time = struct_time(
-                        localtime(
+                    block_time = time.struct_time(
+                        time.localtime(
                             full_block.foliage_transaction_block.timestamp
                             if full_block.foliage_transaction_block
                             else None
diff --git a/chia/cmds/wallet_funcs.py b/chia/cmds/wallet_funcs.py
index b9987d1c42cb..b468b8531edb 100644
--- a/chia/cmds/wallet_funcs.py
+++ b/chia/cmds/wallet_funcs.py
@@ -8,8 +8,10 @@
 
 import aiohttp
 
+from chia.cmds.show import print_connections
 from chia.cmds.units import units
 from chia.rpc.wallet_rpc_client import WalletRpcClient
+from chia.server.outbound_message import NodeType
 from chia.server.start_wallet import SERVICE_NAME
 from chia.types.blockchain_format.sized_bytes import bytes32
 from chia.util.bech32m import encode_puzzle_hash
@@ -471,6 +473,10 @@ async def print_balances(args: dict, wallet_client: WalletRpcClient, fingerprint
         )
         print(f"   -Spendable: {print_balance(balances['spendable_balance'], scale, address_prefix)}")
 
+    print(" ")
+    trusted_peers: Dict = config.get("trusted_peers", {})
+    await print_connections(wallet_client, time, NodeType, trusted_peers)
+
 
 async def get_wallet(wallet_client: WalletRpcClient, fingerprint: int = None) -> Optional[Tuple[WalletRpcClient, int]]:
     if fingerprint is not None:
diff --git a/chia/full_node/full_node.py b/chia/full_node/full_node.py
index 97f3cdebc26e..d0ac3fe2a713 100644
--- a/chia/full_node/full_node.py
+++ b/chia/full_node/full_node.py
@@ -939,10 +939,10 @@ async def validate_block_batches(batch_queue):
                     await peer.close(600)
                     raise ValueError(f"Failed to validate block batch {start_height} to {end_height}")
                 self.log.info(f"Added blocks {start_height} to {end_height}")
-                await self.send_peak_to_wallets()
                 peak = self.blockchain.get_peak()
                 if len(coin_states) > 0 and fork_height is not None:
                     await self.update_wallets(peak.height, fork_height, peak.header_hash, coin_states)
+                await self.send_peak_to_wallets()
                 self.blockchain.clean_block_record(end_height - self.constants.BLOCKS_CACHE_SIZE)
 
         loop = asyncio.get_event_loop()
diff --git a/chia/full_node/lock_queue.py b/chia/full_node/lock_queue.py
index 3f987485c95a..7ff004a77a4e 100644
--- a/chia/full_node/lock_queue.py
+++ b/chia/full_node/lock_queue.py
@@ -31,9 +31,9 @@ class LockQueue:
     Must be created under an asyncio running loop, and close and await_closed should be called.
     """
 
-    def __init__(self, inner_lock: asyncio.Lock):
+    def __init__(self, inner_lock: asyncio.Lock, maxsize=0):
         self._inner_lock: asyncio.Lock = inner_lock
-        self._task_queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
+        self._task_queue: asyncio.PriorityQueue = asyncio.PriorityQueue(maxsize=maxsize)
         self._run_task = asyncio.create_task(self._run())
         self._release_event = asyncio.Event()
 
diff --git a/chia/pools/pool_wallet.py b/chia/pools/pool_wallet.py
index 595db32838f7..e45c03886b3b 100644
--- a/chia/pools/pool_wallet.py
+++ b/chia/pools/pool_wallet.py
@@ -350,7 +350,7 @@ async def create(
 
         p2_puzzle_hash: bytes32 = (await self.get_current_state()).p2_singleton_puzzle_hash
         await self.wallet_state_manager.add_new_wallet(self, self.wallet_info.id, create_puzzle_hashes=False)
-        await self.wallet_state_manager.add_interested_puzzle_hash(p2_puzzle_hash, self.wallet_id, False)
+        await self.wallet_state_manager.add_interested_puzzle_hashes([p2_puzzle_hash], [self.wallet_id], False)
         return self
 
     @staticmethod
diff --git a/chia/util/merkle_set.py b/chia/util/merkle_set.py
index 7f16fa1dd7da..857b8c31d4d1 100644
--- a/chia/util/merkle_set.py
+++ b/chia/util/merkle_set.py
@@ -349,23 +349,23 @@ class SetError(Exception):
     pass
 
 
-def confirm_included(root: Node, val: bytes, proof: bytes32) -> bool:
+def confirm_included(root: bytes32, val: bytes, proof: bytes32) -> bool:
     return confirm_not_included_already_hashed(root, sha256(val).digest(), proof)
 
 
-def confirm_included_already_hashed(root: Node, val: bytes, proof: bytes) -> bool:
+def confirm_included_already_hashed(root: bytes32, val: bytes, proof: bytes) -> bool:
     return _confirm(root, val, proof, True)
 
 
-def confirm_not_included(root: Node, val: bytes, proof: bytes32) -> bool:
+def confirm_not_included(root: bytes32, val: bytes, proof: bytes32) -> bool:
     return confirm_not_included_already_hashed(root, sha256(val).digest(), proof)
 
 
-def confirm_not_included_already_hashed(root: Node, val: bytes, proof: bytes) -> bool:
+def confirm_not_included_already_hashed(root: bytes32, val: bytes, proof: bytes) -> bool:
     return _confirm(root, val, proof, False)
 
 
-def _confirm(root: Node, val: bytes, proof: bytes, expected: bool) -> bool:
+def _confirm(root: bytes32, val: bytes, proof: bytes, expected: bool) -> bool:
     try:
         p = deserialize_proof(proof)
         if p.get_root() != root:
diff --git a/chia/util/network.py b/chia/util/network.py
index 2985505a40fc..3bce97ac5fef 100644
--- a/chia/util/network.py
+++ b/chia/util/network.py
@@ -1,7 +1,8 @@
 import socket
 from ipaddress import ip_address, IPv4Network, IPv6Network
-from typing import Iterable, List, Tuple, Union, Any, Optional
+from typing import Iterable, List, Tuple, Union, Any, Optional, Dict
 from chia.server.outbound_message import NodeType
+from chia.types.blockchain_format.sized_bytes import bytes32
 from chia.types.peer_info import PeerInfo
 from chia.util.ints import uint16
 
@@ -72,3 +73,14 @@ def get_host_addr(host: Union[PeerInfo, str], prefer_ipv6: Optional[bool]) -> st
             return t[4][0]
     # If neither matched preference, just return the first available
     return addrset[0][4][0]
+
+
+def is_trusted_inner(peer_host: str, peer_node_id: bytes32, trusted_peers: Dict, testing: bool) -> bool:
+    if trusted_peers is None:
+        return False
+    if not testing and peer_host == "127.0.0.1":
+        return True
+    if peer_node_id.hex() not in trusted_peers:
+        return False
+
+    return True
diff --git a/chia/wallet/key_val_store.py b/chia/wallet/key_val_store.py
index 3a2a509429d7..1fbdd05a37eb 100644
--- a/chia/wallet/key_val_store.py
+++ b/chia/wallet/key_val_store.py
@@ -3,7 +3,6 @@
 import aiosqlite
 
 from chia.util.db_wrapper import DBWrapper
-from chia.util.streamable import Streamable
 
 
 class KeyValStore:
@@ -47,9 +46,9 @@ async def get_object(self, key: str, object_type: Any) -> Any:
 
         return object_type.from_bytes(row[1])
 
-    async def set_object(self, key: str, obj: Streamable):
+    async def set_object(self, key: str, obj: Any):
         """
-        Adds object to key val store
+        Adds object to key val store. Obj MUST support __bytes__ and bytes() methods.
         """
         async with self.db_wrapper.lock:
             cursor = await self.db_connection.execute(
diff --git a/chia/wallet/rl_wallet/rl_wallet.py b/chia/wallet/rl_wallet/rl_wallet.py
index b18e0372cc04..abfd7126f9b1 100644
--- a/chia/wallet/rl_wallet/rl_wallet.py
+++ b/chia/wallet/rl_wallet/rl_wallet.py
@@ -1,5 +1,4 @@
 # RLWallet is subclass of Wallet
-import asyncio
 import json
 import time
 from dataclasses import dataclass
@@ -324,7 +323,7 @@ async def aggregate_this_coin(self, coin: Coin):
             memos=list(compute_memos(spend_bundle).items()),
         )
 
-        asyncio.create_task(self.push_transaction(tx_record))
+        await self.push_transaction(tx_record)
 
     async def rl_available_balance(self) -> uint64:
         self.rl_coin_record = await self._get_rl_coin_record()
diff --git a/chia/wallet/trade_manager.py b/chia/wallet/trade_manager.py
index 52b2f0d0cb6f..6c8303ce3126 100644
--- a/chia/wallet/trade_manager.py
+++ b/chia/wallet/trade_manager.py
@@ -81,7 +81,7 @@ async def get_trade_by_coin(self, coin: Coin) -> Optional[TradeRecord]:
                 return trade
         return None
 
-    async def coins_of_interest_farmed(self, coin_state: CoinState):
+    async def coins_of_interest_farmed(self, coin_state: CoinState, fork_height: Optional[uint32]):
         """
         If both our coins and other coins in trade got removed that means that trade was successfully executed
         If coins from other side of trade got farmed without ours, that means that trade failed because either someone
@@ -110,7 +110,7 @@ async def coins_of_interest_farmed(self, coin_state: CoinState):
         our_settlement_ids: List[bytes32] = [c.name() for c in our_settlement_payments]
 
         # And get all relevant coin states
-        coin_states = await self.wallet_state_manager.wallet_node.get_coin_state(our_settlement_ids)
+        coin_states = await self.wallet_state_manager.wallet_node.get_coin_state(our_settlement_ids, fork_height)
         assert coin_states is not None
         coin_state_names: List[bytes32] = [cs.coin.name() for cs in coin_states]
 
diff --git a/chia/wallet/util/filter_coin_states.py b/chia/wallet/util/filter_coin_states.py
new file mode 100644
index 000000000000..3d9aaee677ba
--- /dev/null
+++ b/chia/wallet/util/filter_coin_states.py
@@ -0,0 +1,17 @@
+from typing import List
+
+from chia.protocols.wallet_protocol import CoinState
+
+
+def filter_coin_states(all_coins_state: List[CoinState], fork_height: int) -> List[CoinState]:
+    # We only want to apply changes before the fork point, since we are synced to another peer
+    # We are just validating that there is no missing information
+    final_coin_state: List[CoinState] = []
+    for coin_state_entry in all_coins_state:
+        if coin_state_entry.spent_height is not None:
+            if coin_state_entry.spent_height <= fork_height:
+                final_coin_state.append(coin_state_entry)
+        elif coin_state_entry.created_height is not None:
+            if coin_state_entry.created_height <= fork_height:
+                final_coin_state.append(coin_state_entry)
+    return final_coin_state
diff --git a/chia/wallet/util/new_peak_queue.py b/chia/wallet/util/new_peak_queue.py
new file mode 100644
index 000000000000..846ccf814395
--- /dev/null
+++ b/chia/wallet/util/new_peak_queue.py
@@ -0,0 +1,70 @@
+import asyncio
+import dataclasses
+from enum import IntEnum
+from typing import Any, List
+
+from chia.protocols.wallet_protocol import CoinStateUpdate, NewPeakWallet
+from chia.server.ws_connection import WSChiaConnection
+from chia.types.blockchain_format.sized_bytes import bytes32
+
+
+class NewPeakQueueTypes(IntEnum):
+    # Lower number means higher priority in the queue
+    COIN_ID_SUBSCRIPTION = 1
+    PUZZLE_HASH_SUBSCRIPTION = 2
+    FULL_NODE_STATE_UPDATED = 3
+    NEW_PEAK_WALLET = 4
+
+
+@dataclasses.dataclass
+class NewPeakItem:
+    item_type: NewPeakQueueTypes
+    data: Any
+
+    def __lt__(self, other):
+        if self.item_type != other.item_type:
+            return self.item_type < other.item_type
+        if self.item_type in {NewPeakQueueTypes.COIN_ID_SUBSCRIPTION, NewPeakQueueTypes.PUZZLE_HASH_SUBSCRIPTION}:
+            return False  # All subscriptions are equal
+        return self.data[0].height < other.data[0].height
+
+    def __le__(self, other):
+        if self.item_type != other.item_type:
+            return self.item_type < other.item_type
+        if self.item_type in {NewPeakQueueTypes.COIN_ID_SUBSCRIPTION, NewPeakQueueTypes.PUZZLE_HASH_SUBSCRIPTION}:
+            return True  # All subscriptions are equal
+        return self.data[0].height <= other.data[0].height
+
+    def __gt__(self, other):
+        if self.item_type != other.item_type:
+            return self.item_type > other.item_type
+        if self.item_type in {NewPeakQueueTypes.COIN_ID_SUBSCRIPTION, NewPeakQueueTypes.PUZZLE_HASH_SUBSCRIPTION}:
+            return False  # All subscriptions are equal
+        return self.data[0].height > other.data[0].height
+
+    def __ge__(self, other):
+        if self.item_type != other.item_type:
+            return self.item_type > other.item_type
+        if self.item_type in {NewPeakQueueTypes.COIN_ID_SUBSCRIPTION, NewPeakQueueTypes.PUZZLE_HASH_SUBSCRIPTION}:
+            return True  # All subscriptions are equal
+        return self.data[0].height >= other.data[0].height
+
+
+class NewPeakQueue:
+    def __init__(self, inner_queue: asyncio.PriorityQueue):
+        self._inner_queue: asyncio.PriorityQueue = inner_queue
+
+    async def subscribe_to_coin_ids(self, coin_ids: List[bytes32]):
+        await self._inner_queue.put(NewPeakItem(NewPeakQueueTypes.COIN_ID_SUBSCRIPTION, coin_ids))
+
+    async def subscribe_to_puzzle_hashes(self, puzzle_hashes: List[bytes32]):
+        await self._inner_queue.put(NewPeakItem(NewPeakQueueTypes.PUZZLE_HASH_SUBSCRIPTION, puzzle_hashes))
+
+    async def full_node_state_updated(self, coin_state_update: CoinStateUpdate, peer: WSChiaConnection):
+        await self._inner_queue.put(NewPeakItem(NewPeakQueueTypes.FULL_NODE_STATE_UPDATED, (coin_state_update, peer)))
+
+    async def new_peak_wallet(self, new_peak: NewPeakWallet, peer: WSChiaConnection):
+        await self._inner_queue.put(NewPeakItem(NewPeakQueueTypes.NEW_PEAK_WALLET, (new_peak, peer)))
+
+    async def get(self) -> NewPeakItem:
+        return await self._inner_queue.get()
diff --git a/chia/wallet/util/peer_request_cache.py b/chia/wallet/util/peer_request_cache.py
new file mode 100644
index 000000000000..12869ba04f53
--- /dev/null
+++ b/chia/wallet/util/peer_request_cache.py
@@ -0,0 +1,51 @@
+from typing import Dict, Tuple, Any, Optional, List
+
+from chia.protocols.wallet_protocol import CoinState
+from chia.types.blockchain_format.sized_bytes import bytes32
+from chia.types.header_block import HeaderBlock
+from chia.util.ints import uint32
+
+
+class PeerRequestCache:
+    blocks: Dict[uint32, HeaderBlock]
+    block_requests: Dict[Tuple[int, int], Any]
+    ses_requests: Dict[int, Any]
+    states_validated: Dict[bytes32, CoinState]
+
+    def __init__(self):
+        self.blocks = {}
+        self.ses_requests = {}
+        self.block_requests = {}
+        self.states_validated = {}
+
+    def clear_after_height(self, height: int):
+        # Remove any cached item which relates to an event that happened at a height above height.
+        self.blocks = {k: v for k, v in self.blocks.items() if k <= height}
+        self.block_requests = {k: v for k, v in self.block_requests.items() if k[0] <= height and k[1] <= height}
+        self.ses_requests = {k: v for k, v in self.ses_requests.items() if k <= height}
+
+        remove_keys_states: List[bytes32] = []
+        for k4, coin_state in self.states_validated.items():
+            if coin_state.created_height is not None and coin_state.created_height > height:
+                remove_keys_states.append(k4)
+            elif coin_state.spent_height is not None and coin_state.spent_height > height:
+                remove_keys_states.append(k4)
+        for k5 in remove_keys_states:
+            self.states_validated.pop(k5)
+
+
+async def can_use_peer_request_cache(
+    coin_state: CoinState, peer_request_cache: PeerRequestCache, fork_height: Optional[uint32]
+):
+    if coin_state.get_hash() not in peer_request_cache.states_validated:
+        return False
+    if fork_height is None:
+        return True
+    if coin_state.created_height is None and coin_state.spent_height is None:
+        # Performing a reorg
+        return False
+    if coin_state.created_height is not None and coin_state.created_height > fork_height:
+        return False
+    if coin_state.spent_height is not None and coin_state.spent_height > fork_height:
+        return False
+    return True
diff --git a/chia/wallet/util/wallet_sync_utils.py b/chia/wallet/util/wallet_sync_utils.py
index 8a8e5090ce1e..d42313a0d704 100644
--- a/chia/wallet/util/wallet_sync_utils.py
+++ b/chia/wallet/util/wallet_sync_utils.py
@@ -1,6 +1,8 @@
+import logging
 from typing import List, Optional, Tuple, Union, Dict
 
 from chia.consensus.constants import ConsensusConstants
+from chia.protocols import wallet_protocol
 from chia.protocols.wallet_protocol import (
     RequestAdditions,
     RespondAdditions,
@@ -8,17 +10,40 @@
     RejectRemovalsRequest,
     RespondRemovals,
     RequestRemovals,
+    RespondBlockHeader,
 )
+from chia.server.ws_connection import WSChiaConnection
 from chia.types.blockchain_format.coin import hash_coin_list, Coin
 from chia.types.blockchain_format.sized_bytes import bytes32
 from chia.types.full_block import FullBlock
+from chia.types.header_block import HeaderBlock
+from chia.util.ints import uint32
 from chia.util.merkle_set import confirm_not_included_already_hashed, confirm_included_already_hashed, MerkleSet
 
 
+log = logging.getLogger(__name__)
+
+
+async def fetch_last_tx_from_peer(height: uint32, peer: WSChiaConnection) -> Optional[HeaderBlock]:
+    request_height: int = height
+    while True:
+        if request_height == -1:
+            return None
+        request = wallet_protocol.RequestBlockHeader(uint32(request_height))
+        response: Optional[RespondBlockHeader] = await peer.request_block_header(request)
+        if response is not None and isinstance(response, RespondBlockHeader):
+            if response.header_block.is_transaction_block:
+                return response.header_block
+        else:
+            break
+        request_height = request_height - 1
+    return None
+
+
 def validate_additions(
     coins: List[Tuple[bytes32, List[Coin]]],
     proofs: Optional[List[Tuple[bytes32, bytes, Optional[bytes]]]],
-    root,
+    root: bytes32,
 ):
     if proofs is None:
         # Verify root
@@ -77,7 +102,9 @@ def validate_additions(
     return True
 
 
-def validate_removals(coins, proofs, root):
+def validate_removals(
+    coins: List[Tuple[bytes32, Optional[Coin]]], proofs: Optional[List[Tuple[bytes32, bytes]]], root: bytes32
+):
     if proofs is None:
         # If there are no proofs, it means all removals were returned in the response.
         # we must find the ones relevant to our wallets.
@@ -124,7 +151,9 @@ def validate_removals(coins, proofs, root):
     return True
 
 
-async def request_and_validate_removals(peer, height, header_hash, coin_name, removals_root) -> bool:
+async def request_and_validate_removals(
+    peer: WSChiaConnection, height: uint32, header_hash: bytes32, coin_name: bytes32, removals_root: bytes32
+) -> bool:
     removals_request = RequestRemovals(height, header_hash, [coin_name])
 
     removals_res: Optional[Union[RespondRemovals, RejectRemovalsRequest]] = await peer.request_removals(
@@ -132,10 +161,13 @@ async def request_and_validate_removals(peer, height, header_hash, coin_name, re
     )
     if removals_res is None or isinstance(removals_res, RejectRemovalsRequest):
         return False
+    assert removals_res.proofs is not None
     return validate_removals(removals_res.coins, removals_res.proofs, removals_root)
 
 
-async def request_and_validate_additions(peer, height, header_hash, puzzle_hash, additions_root):
+async def request_and_validate_additions(
+    peer: WSChiaConnection, height: uint32, header_hash: bytes32, puzzle_hash: bytes32, additions_root: bytes32
+):
     additions_request = RequestAdditions(height, header_hash, [puzzle_hash])
     additions_res: Optional[Union[RespondAdditions, RejectAdditionsRequest]] = await peer.request_additions(
         additions_request
@@ -143,12 +175,11 @@ async def request_and_validate_additions(peer, height, header_hash, puzzle_hash,
     if additions_res is None or isinstance(additions_res, RejectAdditionsRequest):
         return False
 
-    validated = validate_additions(
+    return validate_additions(
         additions_res.coins,
         additions_res.proofs,
         additions_root,
     )
-    return validated
 
 
 def get_block_challenge(
diff --git a/chia/wallet/wallet_blockchain.py b/chia/wallet/wallet_blockchain.py
index 6b55efbb6b86..cf5c6da1ad81 100644
--- a/chia/wallet/wallet_blockchain.py
+++ b/chia/wallet/wallet_blockchain.py
@@ -24,6 +24,7 @@ class WalletBlockchain(BlockchainInterface):
     _weight_proof_handler: WalletWeightProofHandler
 
     synced_weight_proof: Optional[WeightProof]
+    _finished_sync_up_to: uint32
 
     _peak: Optional[HeaderBlock]
     _height_to_hash: Dict[uint32, bytes32]
@@ -48,6 +49,9 @@ async def create(
         self.CACHE_SIZE = constants.SUB_EPOCH_BLOCKS + 100
         self._weight_proof_handler = weight_proof_handler
         self.synced_weight_proof = await self._basic_store.get_object("SYNCED_WEIGHT_PROOF", WeightProof)
+        self._finished_sync_up_to = await self._basic_store.get_object("FINISHED_SYNC_UP_TO", uint32)
+        if self._finished_sync_up_to is None:
+            self._finished_sync_up_to = uint32(0)
         self._peak = None
         self._peak = await self.get_peak_block()
         self._latest_timestamp = uint64(0)
@@ -156,6 +160,8 @@ async def _rollback_to_height(self, height: int):
         await self._basic_store.remove_object("PEAK_BLOCK")
 
     def get_peak_height(self) -> uint32:
+        # The peak height is the latest height that we know of in the blockchain, it does not mean
+        # that we have downloaded all transactions up to that height.
         if self._peak is None:
             return uint32(0)
         return self._peak.height
@@ -167,13 +173,22 @@ async def set_peak_block(self, block: HeaderBlock, timestamp: Optional[uint64] =
             self._latest_timestamp = timestamp
         elif block.foliage_transaction_block is not None:
             self._latest_timestamp = block.foliage_transaction_block.timestamp
-        log.info(f"Peak set to : {self._peak.height} timestamp: {self._latest_timestamp}")
+        log.info(f"Peak set to: {self._peak.height} timestamp: {self._latest_timestamp}")
 
     async def get_peak_block(self) -> Optional[HeaderBlock]:
         if self._peak is not None:
             return self._peak
         return await self._basic_store.get_object("PEAK_BLOCK", HeaderBlock)
 
+    async def set_finished_sync_up_to(self, height: uint32):
+        await self._basic_store.set_object("FINISHED_SYNC_UP_TO", height)
+
+    async def get_finished_sync_up_to(self):
+        h: Optional[uint32] = await self._basic_store.get_object("FINISHED_SYNC_UP_TO", uint32)
+        if h is None:
+            return uint32(0)
+        return h
+
     def get_latest_timestamp(self) -> uint64:
         return self._latest_timestamp
 
diff --git a/chia/wallet/wallet_node.py b/chia/wallet/wallet_node.py
index e86ae13eed4a..d7d2c9ea2202 100644
--- a/chia/wallet/wallet_node.py
+++ b/chia/wallet/wallet_node.py
@@ -52,9 +52,13 @@
 from chia.util.ints import uint32, uint64
 from chia.util.keychain import KeyringIsLocked, Keychain
 from chia.util.path import mkdir, path_from_root
+from chia.wallet.util.filter_coin_states import filter_coin_states
+from chia.wallet.util.new_peak_queue import NewPeakQueue, NewPeakQueueTypes, NewPeakItem
+from chia.wallet.util.peer_request_cache import PeerRequestCache, can_use_peer_request_cache
 from chia.wallet.util.wallet_sync_utils import (
     request_and_validate_removals,
     request_and_validate_additions,
+    fetch_last_tx_from_peer,
 )
 from chia.wallet.wallet_coin_record import WalletCoinRecord
 from chia.wallet.wallet_state_manager import WalletStateManager
@@ -63,34 +67,6 @@
 from chia.util.profiler import profile_task
 
 
-class PeerRequestCache:
-    blocks: Dict[uint32, HeaderBlock]
-    block_requests: Dict[Tuple[int, int], Any]
-    ses_requests: Dict[int, Any]
-    states_validated: Dict[bytes32, CoinState]
-
-    def __init__(self):
-        self.blocks = {}
-        self.ses_requests = {}
-        self.block_requests = {}
-        self.states_validated = {}
-
-    def clear_after_height(self, height: int):
-        # Remove any cached item which relates to an event that happened at a height above height.
-        self.blocks = {k: v for k, v in self.blocks.items() if k <= height}
-        self.block_requests = {k: v for k, v in self.block_requests.items() if k[0] <= height and k[1] <= height}
-        self.ses_requests = {k: v for k, v in self.ses_requests.items() if k <= height}
-
-        remove_keys_states: List[bytes32] = []
-        for k4, coin_state in self.states_validated.items():
-            if coin_state.created_height is not None and coin_state.created_height > height:
-                remove_keys_states.append(k4)
-            elif coin_state.spent_height is not None and coin_state.spent_height > height:
-                remove_keys_states.append(k4)
-        for k5 in remove_keys_states:
-            self.states_validated.pop(k5)
-
-
 class WalletNode:
     key_config: Dict
     config: Dict
@@ -109,7 +85,11 @@ class WalletNode:
     wallet_peers_initialized: bool
     keychain_proxy: Optional[KeychainProxy]
     wallet_peers: Optional[WalletPeers]
-    race_cache: Dict[uint32, Set[CoinState]]
+    race_cache: Dict[bytes32, Set[CoinState]]
+    race_cache_hashes: List[Tuple[uint32, bytes32]]
+    new_peak_queue: NewPeakQueue
+    _process_new_subscriptions_task: Optional[asyncio.Task]
+    node_peaks: Dict[bytes32, Tuple[uint32, bytes32]]
     validation_semaphore: Optional[asyncio.Semaphore]
     local_node_synced: bool
     new_state_lock: Optional[asyncio.Lock]
@@ -145,14 +125,18 @@ def __init__(
         self.keychain_proxy = None
         self.local_keychain = local_keychain
         self.height_to_time: Dict[uint32, uint64] = {}
-        self.synced_peers: Set[bytes32] = set()
+        self.synced_peers: Set[bytes32] = set()  # Peers that we have long synced to
         self.wallet_peers = None
         self.wallet_peers_initialized = False
         self.valid_wp_cache: Dict[bytes32, Any] = {}
         self.untrusted_caches: Dict[bytes32, Any] = {}
         self.race_cache = {}  # in Untrusted mode wallet might get the state update before receiving the block
+        self.race_cache_hashes = []
+        self._process_new_subscriptions_task = None
+        self.node_peaks = {}
         self.validation_semaphore = None
         self.local_node_synced = False
+        self.LONG_SYNC_THRESHOLD = 200
 
     async def ensure_keychain_proxy(self) -> KeychainProxy:
         if not self.keychain_proxy:
@@ -194,6 +178,9 @@ async def _start(
         self,
         fingerprint: Optional[int] = None,
     ) -> bool:
+        # Makes sure the coin_state_updates get higher priority than new_peak messages
+        self.new_peak_queue = NewPeakQueue(asyncio.PriorityQueue())
+
         self.synced_peers = set()
         private_key = await self.get_key_for_fingerprint(fingerprint)
         if private_key is None:
@@ -242,6 +229,7 @@ async def _start(
 
         self.wallet_state_manager.set_pending_callback(self._pending_tx_handler)
         self._shut_down = False
+        self._process_new_subscriptions_task = asyncio.create_task(self._process_new_subscriptions())
 
         self.sync_event = asyncio.Event()
         if fingerprint is None:
@@ -258,21 +246,17 @@ async def _start(
                 self.wsm_close_task = None
         return True
 
-    async def new_puzzle_hash_created(self, puzzle_hashes: List[bytes32]):
-        if len(puzzle_hashes) == 0:
-            return
-        assert self.server is not None
-        full_nodes: Dict[bytes32, WSChiaConnection] = self.server.connection_by_type.get(NodeType.FULL_NODE, {})
-        for node_id, node in full_nodes.copy().items():
-            asyncio.create_task(self.subscribe_to_phs(puzzle_hashes, node))
-
     def _close(self):
         self.log.info("self._close")
         self.logged_in_fingerprint = None
         self._shut_down = True
 
+        if self._process_new_subscriptions_task is not None:
+            self._process_new_subscriptions_task.cancel()
+
     async def _await_closed(self):
         self.log.info("self._await_closed")
+
         if self.server is not None:
             await self.server.close_all_connections()
         if self.wallet_peers is not None:
@@ -355,6 +339,62 @@ async def _messages_to_resend(self) -> List[Tuple[Message, Set[bytes32]]]:
 
         return messages
 
+    async def _process_new_subscriptions(self):
+        while not self._shut_down:
+            # Here we process four types of messages in the queue, where the first one has higher priority (lower
+            # number in the queue), and priority decreases for each type.
+            peer: Optional[WSChiaConnection] = None
+            item: Optional[NewPeakItem] = None
+            try:
+                peer, item = None, None
+                item = await self.new_peak_queue.get()
+                assert item is not None
+                if item.item_type == NewPeakQueueTypes.COIN_ID_SUBSCRIPTION:
+                    # Subscriptions are the highest priority, because we don't want to process any more peaks or
+                    # state updates until we are sure that we subscribed to everything that we need to. Otherwise,
+                    # we might not be able to process some state.
+                    coin_ids: List[bytes32] = item.data
+                    self.log.info(f"Processing new Coin subscription: {coin_ids}")
+                    for peer in self.server.get_full_node_connections():
+                        coin_states: List[CoinState] = await self.subscribe_to_coin_updates(
+                            coin_ids, peer, True, uint32(0), None
+                        )
+                        if len(coin_states) > 0:
+                            async with self.wallet_state_manager.lock:
+                                await self.receive_state_from_peer(coin_states, peer)
+                elif item.item_type == NewPeakQueueTypes.PUZZLE_HASH_SUBSCRIPTION:
+                    puzzle_hashes: List[bytes32] = item.data
+                    self.log.info(f"Processing new PH subscription: {puzzle_hashes}")
+                    for peer in self.server.get_full_node_connections():
+                        # Puzzle hash subscription
+                        coin_states: List[CoinState] = await self.subscribe_to_phs(
+                            puzzle_hashes, peer, True, uint32(0), None
+                        )
+                        if len(coin_states) > 0:
+                            async with self.wallet_state_manager.lock:
+                                await self.receive_state_from_peer(coin_states, peer)
+                elif item.item_type == NewPeakQueueTypes.FULL_NODE_STATE_UPDATED:
+                    # Note: this can take a while when we have a lot of transactions. We want to process these
+                    # before new_peaks, since new_peak_wallet requires that we first obtain the state for that peak.
+                    request: wallet_protocol.CoinStateUpdate = item.data[0]
+                    peer = item.data[1]
+                    assert peer is not None
+                    await self.state_update_received(request, peer)
+                elif item.item_type == NewPeakQueueTypes.NEW_PEAK_WALLET:
+                    # This can take a VERY long time, because it might trigger a long sync. It is OK if we miss some
+                    # subscriptions or state updates, since all subscriptions and state updates will be handled by
+                    # long_sync (up to the target height).
+                    request: wallet_protocol.NewPeakWallet = item.data[0]
+                    peer = item.data[1]
+                    assert peer is not None
+                    await self.new_peak_wallet(request, peer)
+                else:
+                    assert False
+            except Exception as e:
+                self.log.error(f"Exception handling {item}, {e} {traceback.format_exc()}")
+                if peer is not None:
+                    await peer.close(9999)
+
     def set_server(self, server: ChiaServer):
         self.server = server
         self.initialize_wallet_peers()
@@ -393,6 +433,10 @@ def on_disconnect(self, peer: WSChiaConnection):
 
         if peer.peer_node_id in self.untrusted_caches:
             self.untrusted_caches.pop(peer.peer_node_id)
+        if peer.peer_node_id in self.synced_peers:
+            self.synced_peers.remove(peer.peer_node_id)
+        if peer.peer_node_id in self.node_peaks:
+            self.node_peaks.pop(peer.peer_node_id)
 
     async def on_connect(self, peer: WSChiaConnection):
         if self.wallet_state_manager is None:
@@ -406,7 +450,10 @@ async def on_connect(self, peer: WSChiaConnection):
         if not trusted and self.local_node_synced:
             await peer.close()
 
-        self.log.info(f"Connected peer {peer} is {trusted}")
+        if peer.peer_node_id in self.synced_peers:
+            self.synced_peers.remove(peer.peer_node_id)
+
+        self.log.info(f"Connected peer {peer.get_peer_info()} is trusted: {trusted}")
         messages_peer_ids = await self._messages_to_resend()
         self.wallet_state_manager.state_changed("add_connection")
         for msg, peer_ids in messages_peer_ids:
@@ -415,67 +462,107 @@ async def on_connect(self, peer: WSChiaConnection):
             await peer.send_message(msg)
 
         if self.wallet_peers is not None:
-            asyncio.create_task(self.wallet_peers.on_connect(peer))
+            await self.wallet_peers.on_connect(peer)
 
-    async def trusted_sync(self, full_node: WSChiaConnection):
+    async def long_sync(
+        self,
+        target_height: uint32,
+        full_node: WSChiaConnection,
+        syncing: bool = False,
+        fork_height: Optional[int] = None,
+    ):
         """
-        Performs a one-time sync with each trusted peer, subscribing to interested puzzle hashes and coin ids.
+        Sync algorithm:
+        - Download and verify weight proof (if not trusted)
+        - The minimum height is finished_sync_up_to - 256, roll back anything after it (if syncing)
+        - Subscribe to all puzzle_hashes over and over until there are no more updates
+        - Subscribe to all coin_ids over and over until there are no more updates
+        - syncing=False means that we are just double-checking with this peer to make sure we don't have any
+          missing transactions, so we don't need to rollback
         """
-        self.log.info("Starting trusted sync")
+
+        trusted: bool = self.is_trusted(full_node)
+        self.log.info(f"Starting sync trusted: {trusted} to peer {full_node.peer_host}")
         assert self.wallet_state_manager is not None
-        self.wallet_state_manager.set_sync_mode(True)
         start_time = time.time()
-        current_height: uint32 = self.wallet_state_manager.blockchain.get_peak_height()
-        request_height: uint32 = uint32(max(0, current_height - 1000))
+        # This is the greatest height where we have downloaded all transactions before it, and we know it's valid
+        current_height: uint32 = await self.wallet_state_manager.blockchain.get_finished_sync_up_to()
+        min_height: uint32 = uint32(max(0, current_height - 256))
+
+        if syncing:
+            await self.wallet_state_manager.reorg_rollback(min_height)
+            self.rollback_request_caches(min_height)
+            await self.update_ui()
 
-        already_checked: Set[bytes32] = set()
+        already_checked_ph: Set[bytes32] = set()
         continue_while: bool = True
+        all_puzzle_hashes: List[bytes32] = await self.get_puzzle_hashes_to_subscribe()
         while continue_while:
             # Get all phs from puzzle store
-            all_puzzle_hashes: List[bytes32] = await self.get_puzzle_hashes_to_subscribe()
-            to_check: List[bytes32] = []
-            for ph in all_puzzle_hashes:
-                if ph in already_checked:
-                    continue
-                else:
-                    to_check.append(ph)
-                    already_checked.add(ph)
-                    if len(to_check) == 1000:
-                        break
-
-            await self.subscribe_to_phs(to_check, full_node, request_height)
+            ph_chunks: List[List[bytes32]] = chunks(all_puzzle_hashes, 1000)
+            for chunk in ph_chunks:
+                ph_update_res: List[CoinState] = await self.subscribe_to_phs(
+                    [p for p in chunk if p not in already_checked_ph], full_node, syncing, min_height, fork_height
+                )
+                await self.receive_state_from_peer(ph_update_res, full_node)
+                already_checked_ph.update(chunk)
 
             # Check if new puzzle hashed have been created
-            check_again = await self.get_puzzle_hashes_to_subscribe()
             await self.wallet_state_manager.create_more_puzzle_hashes()
+            all_puzzle_hashes = await self.get_puzzle_hashes_to_subscribe()
+            continue_while = False
+            for ph in all_puzzle_hashes:
+                if ph not in already_checked_ph:
+                    continue_while = True
+                    break
+        self.log.info(f"Successfully subscribed and updated {len(already_checked_ph)} puzzle hashes")
+
+        continue_while = False
+        all_coin_ids: List[bytes32] = await self.get_coin_ids_to_subscribe(min_height)
+        already_checked_coin_ids: Set[bytes32] = set()
+        while continue_while:
+            one_k_chunks = chunks(all_coin_ids, 1000)
+            for chunk in one_k_chunks:
+                c_update_res: List[CoinState] = await self.subscribe_to_coin_updates(
+                    chunk, full_node, syncing, min_height, fork_height
+                )
+                await self.receive_state_from_peer(c_update_res, full_node)
+                already_checked_coin_ids.update(chunk)
 
+            all_coin_ids = await self.get_coin_ids_to_subscribe(min_height)
             continue_while = False
-            for ph in check_again:
-                if ph not in already_checked:
+            for coin_id in all_coin_ids:
+                if coin_id not in already_checked_coin_ids:
                     continue_while = True
                     break
+        self.log.info(f"Successfully subscribed and updated {len(already_checked_coin_ids)} coin ids")
 
-        all_coins: Set[WalletCoinRecord] = await self.wallet_state_manager.coin_store.get_coins_to_check(request_height)
-        all_coin_names: List[bytes32] = [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())
+        if target_height > await self.wallet_state_manager.blockchain.get_finished_sync_up_to():
+            await self.wallet_state_manager.blockchain.set_finished_sync_up_to(target_height)
+
+        if trusted:
+            self.local_node_synced = True
+
+        self.wallet_state_manager.state_changed("new_block")
+
+        self.synced_peers.add(full_node.peer_node_id)
+        await self.update_ui()
 
-        one_k_chunks = chunks(all_coin_names, 1000)
-        for chunk in one_k_chunks:
-            await self.subscribe_to_coin_updates(chunk, full_node, request_height)
-        self.wallet_state_manager.set_sync_mode(False)
         end_time = time.time()
         duration = end_time - start_time
-        self.log.info(f"Trusted sync duration was: {duration}")
-        # Refresh wallets
-        for wallet_id, wallet in self.wallet_state_manager.wallets.items():
-            self.wallet_state_manager.state_changed("coin_removed", wallet_id)
-            self.wallet_state_manager.state_changed("coin_added", wallet_id)
-        self.synced_peers.add(full_node.peer_node_id)
+        self.log.info(f"Sync (trusted: {trusted}) duration was: {duration}")
 
     async def receive_state_from_peer(
-        self, items: List[CoinState], peer: WSChiaConnection, fork_height: Optional[uint32], height: Optional[uint32]
+        self,
+        items: List[CoinState],
+        peer: WSChiaConnection,
+        fork_height: Optional[uint32] = None,
+        height: Optional[uint32] = None,
+        header_hash: Optional[bytes32] = None,
     ):
+        # Adds the state to the wallet state manager. If the peer is trusted, we do not validate. If the peer is
+        # untrusted we do, but we might not add the state, since we need to receive the new_peak message as well.
+
         assert self.wallet_state_manager is not None
         trusted = self.is_trusted(peer)
         # Validate states in parallel, apply serial
@@ -487,6 +574,9 @@ async def receive_state_from_peer(
         # 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)
+        cache: PeerRequestCache = self.get_cache_for_peer(peer)
+        if fork_height is not None:
+            cache.clear_after_height(fork_height)
 
         all_tasks = []
 
@@ -498,24 +588,18 @@ async def receive_and_validate(inner_state: CoinState, inner_idx: int):
                 # if height is not None:
                 async with self.validation_semaphore:
                     try:
+                        if header_hash is not None:
+                            assert height is not None
+                            self.add_state_to_race_cache(header_hash, height, inner_state)
                         if trusted:
                             valid = True
                         else:
-                            valid = await self.validate_received_state_from_peer(
-                                inner_state, peer, self.get_cache_for_peer(peer)
-                            )
+                            valid = await self.validate_received_state_from_peer(inner_state, peer, cache, fork_height)
                         if valid:
-                            self.log.info(f"new coin state received ({inner_idx} / {len(items)})")
+                            self.log.info(f"new coin state received ({inner_idx + 1} / {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)
+                                await self.wallet_state_manager.new_coin_state([inner_state], peer, fork_height)
                     except Exception as e:
                         tb = traceback.format_exc()
                         self.log.error(f"Exception while adding state: {e} {tb}")
@@ -527,60 +611,58 @@ async def receive_and_validate(inner_state: CoinState, inner_idx: int):
                 await asyncio.sleep(2)
 
         await asyncio.gather(*all_tasks)
+        await self.update_ui()
 
-    async def subscribe_to_phs(self, puzzle_hashes: List[bytes32], peer: WSChiaConnection, height=uint32(0)):
+    async def subscribe_to_phs(
+        self,
+        puzzle_hashes: List[bytes32],
+        peer: WSChiaConnection,
+        syncing: bool,
+        min_height: uint32,
+        fork_height: Optional[int],
+    ) -> List[CoinState]:
         """
-        Tell full nodes that we are interested in puzzle hashes, and for trusted connections, add the new coin state
-        for the puzzle hashes.
+        Tells full nodes that we are interested in puzzle hashes, and returns the response.
         """
         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)
-        if all_state is None:
-            return
-        await self.receive_state_from_peer(all_state.coin_states, peer, None, None)
+        trusted: bool = self.is_trusted(peer)
+        msg = wallet_protocol.RegisterForPhUpdates(puzzle_hashes, min_height)
+        all_coins_state: Optional[RespondToPhUpdates] = await peer.register_interest_in_puzzle_hash(msg)
+
+        final_coin_state: List[CoinState] = []
+        if all_coins_state is not None:
+            if not trusted and syncing is not None and not syncing:
+                assert fork_height is not None
+                final_coin_state = filter_coin_states(all_coins_state.coin_states, fork_height)
+            else:
+                final_coin_state = all_coins_state.coin_states
+        return final_coin_state
 
-    async def subscribe_to_coin_updates(self, coin_names: List[bytes32], peer: WSChiaConnection, height=uint32(0)):
+    async def subscribe_to_coin_updates(
+        self,
+        coin_names: List[bytes32],
+        peer: WSChiaConnection,
+        syncing: Optional[bool],
+        min_height: uint32,
+        fork_height: Optional[int],
+    ) -> List[CoinState]:
         """
-        Tell full nodes that we are interested in coin ids, and for trusted connections, add the new coin state
-        for the coin changes.
+        Tells full nodes that we are interested in coin ids, and returns the response.
         """
-        msg = wallet_protocol.RegisterForCoinUpdates(coin_names, height)
-        all_coins_state: Optional[RespondToCoinUpdates] = await peer.register_interest_in_coin(msg)
-        if all_coins_state is not None and self.is_trusted(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
-    ) -> List[CoinState]:
-        assert self.server is not None
-        all_nodes = self.server.connection_by_type[NodeType.FULL_NODE]
-        if len(all_nodes.keys()) == 0:
-            raise ValueError("Not connected to the full node")
+        trusted: bool = self.is_trusted(peer)
 
-        # Use supplied if provided, prioritize trusted otherwise
-        if peer is None:
-            for node in list(all_nodes.values()):
-                if self.is_trusted(node):
-                    peer = node
-                    break
-            if peer is None:
-                peer = list(all_nodes.values())[0]
+        msg = wallet_protocol.RegisterForCoinUpdates(coin_names, min_height)
 
-        assert peer is not None
-        msg = wallet_protocol.RegisterForCoinUpdates(coin_names, uint32(0))
-        coin_state: Optional[RespondToCoinUpdates] = await peer.register_interest_in_coin(msg)
-        assert coin_state is not None
-
-        if not self.is_trusted(peer):
-            valid_list = []
-            for coin in coin_state.coin_states:
-                valid = await self.validate_received_state_from_peer(coin, peer, self.get_cache_for_peer(peer))
-                if valid:
-                    valid_list.append(coin)
-            return valid_list
+        all_coins_state: Optional[RespondToCoinUpdates] = await peer.register_interest_in_coin(msg)
 
-        return coin_state.coin_states
+        final_coin_state: List[CoinState] = []
+        if all_coins_state is not None:
+            if not trusted and syncing is not None and not syncing:
+                assert fork_height is not None
+                final_coin_state = filter_coin_states(all_coins_state.coin_states, fork_height)
+            else:
+                final_coin_state = all_coins_state.coin_states
+        return final_coin_state
 
     async def get_coins_with_puzzle_hash(self, puzzle_hash) -> List[CoinState]:
         assert self.wallet_state_manager is not None
@@ -594,70 +676,72 @@ async def get_coins_with_puzzle_hash(self, puzzle_hash) -> List[CoinState]:
         assert coin_state is not None
         return coin_state.coin_states
 
-    def is_trusted(self, peer):
+    async def is_peer_synced(self, peer: WSChiaConnection, height: uint32, request_time: uint64) -> Optional[uint64]:
+        request = wallet_protocol.RequestBlockHeader(height)
+        header_response: Optional[RespondBlockHeader] = await peer.request_block_header(request)
+        assert header_response is not None
+
+        # Get last timestamp
+        last_tx: Optional[HeaderBlock] = await fetch_last_tx_from_peer(height, peer)
+        latest_timestamp: Optional[uint64] = None
+        if last_tx is not None:
+            assert last_tx.foliage_transaction_block is not None
+            latest_timestamp = last_tx.foliage_transaction_block.timestamp
+
+        # Return None if not synced
+        if latest_timestamp is None or self.config["testing"] is False and latest_timestamp < request_time - 600:
+            return None
+        return latest_timestamp
+
+    def is_trusted(self, peer) -> bool:
+        assert self.server is not None
         return self.server.is_trusted_peer(peer, self.config["trusted_peers"])
 
-    def add_state_to_race_cache(self, height: uint32, coin_state: CoinState):
-        if height not in self.race_cache:
-            self.race_cache[height] = set()
-        self.race_cache[height].add(coin_state)
+    def add_state_to_race_cache(self, header_hash: bytes32, height: uint32, coin_state: CoinState) -> None:
+        # Clears old state that is no longer relevant
+        delete_threshold = 100
+        for rc_height, rc_hh in self.race_cache_hashes:
+            if height - delete_threshold >= rc_height:
+                self.race_cache.pop(rc_hh)
+        self.race_cache_hashes = [
+            (rc_height, rc_hh) for rc_height, rc_hh in self.race_cache_hashes if height - delete_threshold < rc_height
+        ]
+
+        if header_hash not in self.race_cache:
+            self.race_cache[header_hash] = set()
+        self.race_cache[header_hash].add(coin_state)
 
-    async def state_update_received(self, request: wallet_protocol.CoinStateUpdate, peer: WSChiaConnection):
+    async def state_update_received(self, request: wallet_protocol.CoinStateUpdate, peer: WSChiaConnection) -> None:
+        # This gets called every time there is a new coin or puzzle hash change in the DB
+        # that is of interest to this wallet. It is not guaranteed to come for every height. This message is guaranteed
+        # to come before the corresponding new_peak for each height. We handle this differently for trusted and
+        # untrusted peers. For trusted, we always process the state, and we process reorgs as well.
         assert self.wallet_state_manager is not None
         assert self.server is not None
 
-        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()
+        async with self.wallet_state_manager.lock:
+            await self.receive_state_from_peer(
+                request.items,
+                peer,
+                request.fork_height,
+                request.height,
+                request.peak_hash,
+            )
 
-    def get_full_node_peer(self):
+    def get_full_node_peer(self) -> Optional[WSChiaConnection]:
+        assert self.server is not None
         nodes = self.server.get_full_node_connections()
         if len(nodes) > 0:
             return nodes[0]
         else:
             return None
 
-    async def last_local_tx_block(self, header_hash: bytes32) -> Optional[BlockRecord]:
-        assert self.wallet_state_manager is not None
-        current_hash = header_hash
-        while True:
-            if self.wallet_state_manager.blockchain.contains_block(current_hash):
-                block = self.wallet_state_manager.blockchain.try_block_record(current_hash)
-                if block is None:
-                    return None
-                if block.is_transaction_block:
-                    return block
-                if block.prev_transaction_block_hash is None:
-                    return None
-                current_hash = block.prev_transaction_block_hash
-            else:
-                break
-        return None
-
-    async def fetch_last_tx_from_peer(self, height: uint32, peer: WSChiaConnection) -> Optional[HeaderBlock]:
-        request_height = height
-        while True:
-            if request_height == 0:
-                return None
-            request = wallet_protocol.RequestBlockHeader(request_height)
-            response: Optional[RespondBlockHeader] = await peer.request_block_header(request)
-            if response is not None and isinstance(response, RespondBlockHeader):
-                if response.header_block.is_transaction_block:
-                    return response.header_block
-            else:
-                break
-            request_height = uint32(request_height - 1)
-        return None
-
     async def disconnect_and_stop_wpeers(self):
+        # Close connection of non trusted peers
         if len(self.server.get_full_node_connections()) > 1:
             for peer in self.server.get_full_node_connections():
                 if not self.is_trusted(peer):
-                    asyncio.create_task(peer.close())
+                    await peer.close()
 
         if self.wallet_peers is not None:
             await self.wallet_peers.ensure_is_closed()
@@ -680,209 +764,159 @@ async def get_timestamp_for_height(self, height: uint32) -> uint64:
                 ):
                     self.height_to_time[height] = block.foliage_transaction_block.timestamp
                     return block.foliage_transaction_block.timestamp
-
-        peer = self.get_full_node_peer()
-        assert peer is not None
-        curr_height: uint32 = height
-        while True:
-            request = wallet_protocol.RequestBlockHeader(curr_height)
-            response: Optional[RespondBlockHeader] = await peer.request_block_header(request)
-            if response is None or not isinstance(response, RespondBlockHeader):
-                raise ValueError(f"Invalid response from {peer}, {response}")
-            if response.header_block.foliage_transaction_block is not None:
-                self.height_to_time[height] = response.header_block.foliage_transaction_block.timestamp
-                return response.header_block.foliage_transaction_block.timestamp
-            curr_height = uint32(curr_height - 1)
+        peer: Optional[WSChiaConnection] = self.get_full_node_peer()
+        if peer is None:
+            raise ValueError("Cannot fetch timestamp, no peers")
+        last_tx_block: Optional[HeaderBlock] = await fetch_last_tx_from_peer(height, peer)
+        if last_tx_block is None:
+            raise ValueError(f"Error fetching blocks from peer {peer.get_peer_info()}")
+        assert last_tx_block.foliage_transaction_block is not None
+        return last_tx_block.foliage_transaction_block.timestamp
 
     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())
-
+        self.log.debug(f"New peak wallet.. {peak.height} {peer.get_peer_info()}")
         if self.wallet_state_manager is None:
             # When logging out of wallet
             return
-        if self.is_trusted(peer):
-            request = wallet_protocol.RequestBlockHeader(peak.height)
-            header_response: Optional[RespondBlockHeader] = await peer.request_block_header(request)
-            assert header_response is not None
-
-            # Get last timestamp
-            last_tx: Optional[HeaderBlock] = await self.fetch_last_tx_from_peer(peak.height, peer)
-            latest_timestamp: Optional[uint64] = None
-            if last_tx is not None:
-                assert last_tx.foliage_transaction_block is not None
-                latest_timestamp = last_tx.foliage_transaction_block.timestamp
-
-            # Ignore if not synced
-            if latest_timestamp is None or self.config["testing"] is False and latest_timestamp < request_time - 600:
-                return
-
-            # Disconnect from all untrusted peers if our local node is trusted and synced
-            await self.disconnect_and_stop_wpeers()
-
-            async with self.new_peak_lock:
-                async with self.wallet_state_manager.lock:
-                    # Sync to trusted node
-                    self.local_node_synced = True
-                    if peer.peer_node_id not in self.synced_peers:
-                        await self.trusted_sync(peer)
+        assert self.server is not None
+        request_time = uint64(int(time.time()))
+        trusted: bool = self.is_trusted(peer)
+        peak_hb: Optional[HeaderBlock] = await self.wallet_state_manager.blockchain.get_peak_block()
+        if peak_hb is not None and peak.weight < peak_hb.weight:
+            # Discards old blocks, but accepts blocks that are equal in weight to peak
+            return
 
-                    await self.wallet_state_manager.blockchain.set_peak_block(
-                        header_response.header_block, latest_timestamp
-                    )
+        latest_timestamp: Optional[uint64] = await self.is_peer_synced(peer, peak.height, request_time)
+        if latest_timestamp is None:
+            if trusted:
+                self.log.debug(f"Trusted peer {peer.get_peer_info()} is not synced.")
+                return
+            else:
+                self.log.warning(f"Non-trusted peer {peer.get_peer_info()} is not synced, disconnecting")
+                await peer.close(120)
+                return
+        request = wallet_protocol.RequestBlockHeader(peak.height)
+        response: Optional[RespondBlockHeader] = await peer.request_block_header(request)
+        if response is None:
+            self.log.warning(f"Peer {peer.get_peer_info()} did not respond in time.")
+            await peer.close(120)
+            return
+        header_block: HeaderBlock = response.header_block
 
-                    self.wallet_state_manager.state_changed("new_block")
+        if self.is_trusted(peer):
+            async with self.wallet_state_manager.lock:
+                await self.wallet_state_manager.blockchain.set_peak_block(header_block, latest_timestamp)
+                # Disconnect from all untrusted peers if our local node is trusted and synced
+                await self.disconnect_and_stop_wpeers()
+
+                # Sync to trusted node if we haven't done so yet. As long as we have synced once (and not
+                # disconnected), we assume that the full node will continue to give us state updates, so we do
+                # not need to resync.
+                if peer.peer_node_id not in self.synced_peers:
+                    await self.long_sync(peak.height, peer, True, None)
                     self.wallet_state_manager.set_sync_mode(False)
         else:
-            async with self.new_peak_lock:
-                request = wallet_protocol.RequestBlockHeader(peak.height)
-                response: Optional[RespondBlockHeader] = await peer.request_block_header(request)
-                if response is None or not isinstance(response, RespondBlockHeader) or response.header_block is None:
-                    self.log.debug(f"bad peak response from peer {response}, perhaps connection was closed")
-                    return
-                peak_block = response.header_block
-                current_peak: Optional[HeaderBlock] = await self.wallet_state_manager.blockchain.get_peak_block()
-                if current_peak is not None and peak_block.weight < current_peak.weight:
-                    if peak_block.height < current_peak.height - 20:
-                        await peer.close(120)
-                    return
-
-                # don't sync if full node is not synced it self, since we want to fully sync to a few peers
-                if (
-                    not response.header_block.is_transaction_block
-                    and current_peak is not None
-                    and peak_block.prev_header_hash == current_peak.header_hash
-                ):
-                    # This block is after our peak, so we don't need to check if node is synced
-                    pass
-                else:
-                    tx_timestamp = None
-                    if not response.header_block.is_transaction_block:
-                        last_tx_block = None
-                        # Try local first
-                        last_block_record = await self.last_local_tx_block(response.header_block.prev_header_hash)
-                        if last_block_record is not None:
-                            tx_timestamp = last_block_record.timestamp
-                        else:
-                            last_tx_block = await self.fetch_last_tx_from_peer(response.header_block.height, peer)
-                            if last_tx_block is not None:
-                                assert last_tx_block.foliage_transaction_block is not None
-                                tx_timestamp = last_tx_block.foliage_transaction_block.timestamp
-                    else:
-                        last_tx_block = response.header_block
-                        assert last_tx_block.foliage_transaction_block is not None
-                        tx_timestamp = last_tx_block.foliage_transaction_block.timestamp
-
-                    if tx_timestamp is None:
-                        return None
-
-                    if self.config["testing"] is False and tx_timestamp < request_time - 600:
-                        # Full node not synced, don't sync to it
-                        self.log.info("Peer we connected to is not fully synced, dropping connection...")
-                        await peer.close()
-                        return
-
-                long_sync_threshold = 200
-                far_behind: bool = (
-                    peak.height - self.wallet_state_manager.blockchain.get_peak_height() > long_sync_threshold
-                )
-
-                # check if claimed peak is heavier or same as our current peak
-                # if we haven't synced fully to this peer sync again
-                if (
-                    peer.peer_node_id not in self.synced_peers or far_behind
-                ) and peak.height >= self.constants.WEIGHT_PROOF_RECENT_BLOCKS:
-                    syncing = False
-                    if far_behind or len(self.synced_peers) == 0:
-                        syncing = True
-                        self.wallet_state_manager.set_sync_mode(True)
-                    try:
-                        (
-                            valid_weight_proof,
-                            weight_proof,
-                            summaries,
-                            block_records,
-                        ) = await self.fetch_and_validate_the_weight_proof(peer, response.header_block)
-                        if valid_weight_proof is False:
-                            if syncing:
-                                self.wallet_state_manager.set_sync_mode(False)
-                            await peer.close()
-                            return
-                        assert weight_proof is not None
-                        old_proof = self.wallet_state_manager.blockchain.synced_weight_proof
-                        curr_peak = await self.wallet_state_manager.blockchain.get_peak_block()
-                        fork_point = 0
-                        if curr_peak is not None:
-                            fork_point = max(0, curr_peak.height - 32)
-
-                        if old_proof is not None:
-                            wp_fork_point = self.wallet_state_manager.weight_proof_handler.get_fork_point(
-                                old_proof, weight_proof
-                            )
-                            if wp_fork_point != 0:
-                                fork_point = wp_fork_point
-
-                        await self.wallet_state_manager.blockchain.new_weight_proof(weight_proof, block_records)
-                        if syncing:
-                            async with self.wallet_state_manager.lock:
-                                await self.untrusted_sync_to_peer(peer, syncing, fork_point)
-                        else:
-                            await self.untrusted_sync_to_peer(peer, syncing, fork_point)
-                        if (
-                            self.wallet_state_manager.blockchain.synced_weight_proof is None
-                            or weight_proof.recent_chain_data[-1].weight
-                            > self.wallet_state_manager.blockchain.synced_weight_proof.recent_chain_data[-1].weight
-                        ):
-                            await self.wallet_state_manager.blockchain.new_weight_proof(weight_proof, block_records)
-
-                        self.synced_peers.add(peer.peer_node_id)
+            far_behind: bool = (
+                peak.height - self.wallet_state_manager.blockchain.get_peak_height() > self.LONG_SYNC_THRESHOLD
+            )
 
-                        self.wallet_state_manager.state_changed("new_block")
-                        self.wallet_state_manager.set_sync_mode(False)
-                        await self.update_ui()
-                    except Exception:
+            # check if claimed peak is heavier or same as our current peak
+            # if we haven't synced fully to this peer sync again
+            if (
+                peer.peer_node_id not in self.synced_peers or far_behind
+            ) and peak.height >= self.constants.WEIGHT_PROOF_RECENT_BLOCKS:
+                syncing = False
+                if far_behind or len(self.synced_peers) == 0:
+                    syncing = True
+                    self.wallet_state_manager.set_sync_mode(True)
+                try:
+                    (
+                        valid_weight_proof,
+                        weight_proof,
+                        summaries,
+                        block_records,
+                    ) = await self.fetch_and_validate_the_weight_proof(peer, response.header_block)
+                    if valid_weight_proof is False:
                         if syncing:
                             self.wallet_state_manager.set_sync_mode(False)
-                        tb = traceback.format_exc()
-                        self.log.error(f"Error syncing to {peer.get_peer_info()} {tb}")
                         await peer.close()
                         return
+                    assert weight_proof is not None
+                    old_proof = self.wallet_state_manager.blockchain.synced_weight_proof
+                    fork_point = 0
+                    if old_proof is not None:
+                        fork_point = self.wallet_state_manager.weight_proof_handler.get_fork_point(
+                            old_proof, weight_proof
+                        )
+
+                    await self.wallet_state_manager.blockchain.new_weight_proof(weight_proof, block_records)
+                    if syncing:
+                        async with self.wallet_state_manager.lock:
+                            await self.long_sync(peak.height, peer, syncing, fork_point)
+                    else:
+                        await self.long_sync(peak.height, peer, syncing, fork_point)
+                    if (
+                        self.wallet_state_manager.blockchain.synced_weight_proof is None
+                        or weight_proof.recent_chain_data[-1].weight
+                        > self.wallet_state_manager.blockchain.synced_weight_proof.recent_chain_data[-1].weight
+                    ):
+                        await self.wallet_state_manager.blockchain.new_weight_proof(weight_proof, block_records)
+                except Exception:
                     if syncing:
                         self.wallet_state_manager.set_sync_mode(False)
+                    tb = traceback.format_exc()
+                    self.log.error(f"Error syncing to {peer.get_peer_info()} {tb}")
+                    await peer.close()
+                    return
+                if syncing:
+                    self.wallet_state_manager.set_sync_mode(False)
+
+            else:
+                # This is the (untrusted) case where we already synced and are not too far behind. Here we just
+                # fetch one by one.
+                async with self.wallet_state_manager.lock:
+                    backtrack_fork_height = await self.wallet_short_sync_backtrack(header_block, peer)
 
-                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
+                        # Edge case, this happens when the peak < WEIGHT_PROOF_RECENT_BLOCKS
+                        # we still want to subscribe for all phs and coins.
                         # (Hints are not in filter)
-                        await self.untrusted_subscribe_to_puzzle_hashes(peer, self.get_cache_for_peer(peer))
+                        all_coin_ids: List[bytes32] = await self.get_coin_ids_to_subscribe(uint32(0))
+                        phs: List[bytes32] = await self.get_puzzle_hashes_to_subscribe()
+                        ph_updates: List[CoinState] = await self.subscribe_to_phs(phs, peer, True, uint32(0), None)
+                        coin_updates: List[CoinState] = await self.subscribe_to_coin_updates(
+                            all_coin_ids, peer, True, uint32(0), None
+                        )
+                        peer_new_peak_height, peer_new_peak_hash = self.node_peaks[peer.peer_node_id]
+                        await self.receive_state_from_peer(
+                            ph_updates + coin_updates,
+                            peer,
+                            height=peer_new_peak_height,
+                            header_hash=peer_new_peak_hash,
+                        )
                         self.synced_peers.add(peer.peer_node_id)
 
-                    if peak_block.height in self.race_cache:
-                        for state in self.race_cache[peak_block.height]:
-                            valid = await self.validate_received_state_from_peer(
-                                state, peer, self.get_cache_for_peer(peer)
-                            )
-                            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
+                    # For every block, we need to apply the cache from race_cache
+                    for potential_height in range(backtrack_fork_height + 1, peak.height + 1):
+                        header_hash = self.wallet_state_manager.blockchain.height_to_hash(uint32(potential_height))
+                        if header_hash in self.race_cache:
+                            self.log.debug(f"Receiving race state: {self.race_cache[header_hash]}")
+                            await self.receive_state_from_peer(list(self.race_cache[header_hash]), peer)
+
                     self.wallet_state_manager.set_sync_mode(False)
                     self.wallet_state_manager.state_changed("new_block")
+                    self.log.info(f"Finished processing new peak of {peak.height}")
 
+        if (
+            peer.peer_node_id in self.synced_peers
+            and peak.height > await self.wallet_state_manager.blockchain.get_finished_sync_up_to()
+        ):
+            await self.wallet_state_manager.blockchain.set_finished_sync_up_to(peak.height)
         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:
+    async def wallet_short_sync_backtrack(self, header_block: HeaderBlock, peer: WSChiaConnection) -> int:
         assert self.wallet_state_manager is not None
+        peak: Optional[HeaderBlock] = await self.wallet_state_manager.blockchain.get_peak_block()
 
         top = header_block
         blocks = [top]
@@ -903,14 +937,12 @@ async def wallet_short_sync_backtrack(self, header_block: HeaderBlock, peer) ->
             fork_height = top.height - 1
 
         blocks.reverse()
-
         # 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()
+            await self.update_ui()
         self.rollback_request_caches(fork_height)
 
         if peak is not None:
@@ -975,125 +1007,31 @@ async def get_puzzle_hashes_to_subscribe(self) -> List[bytes32]:
         all_puzzle_hashes.extend(interested_puzzle_hashes)
         return all_puzzle_hashes
 
-    async def untrusted_subscribe_to_puzzle_hashes(
-        self,
-        peer: WSChiaConnection,
-        peer_request_cache: Optional[PeerRequestCache],
-    ):
+    async def get_coin_ids_to_subscribe(self, min_height: uint32) -> List[bytes32]:
         assert self.wallet_state_manager is not None
-        already_checked = set()
-        continue_while = True
-        while continue_while:
-            all_puzzle_hashes = await self.get_puzzle_hashes_to_subscribe()
-            to_check = []
-            for ph in all_puzzle_hashes:
-                if ph in already_checked:
-                    continue
-                else:
-                    to_check.append(ph)
-                    already_checked.add(ph)
-                    if len(to_check) == 1000:
-                        break
-            msg = wallet_protocol.RegisterForPhUpdates(to_check, uint32(0))
-            all_state: Optional[RespondToPhUpdates] = await peer.register_interest_in_puzzle_hash(msg)
-            assert all_state is not 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()
-
-            continue_while = False
-            for ph in check_again:
-                if ph not in already_checked:
-                    continue_while = True
-                    break
-
-    async def untrusted_sync_to_peer(self, peer: WSChiaConnection, syncing: bool, fork_height: int):
-        assert self.wallet_state_manager is not None
-        # If new weight proof is higher than the old one, rollback to the fork point and than apply new coin_states
-        self.log.info(f"Starting untrusted sync to: {peer.get_peer_info()}, syncing: {syncing}, fork at: {fork_height}")
-        if syncing:
-            self.log.info(f"Rollback for {fork_height}")
-            await self.wallet_state_manager.reorg_rollback(fork_height)
-
-        start_time: float = time.time()
-        peer_request_cache: PeerRequestCache = self.get_cache_for_peer(peer)
-        self.untrusted_caches[peer.peer_node_id] = peer_request_cache
-        # 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, peer_request_cache)
-        self.log.debug("End untrusted_subscribe_to_puzzle_hashes  ")
-
-        checked_all_coins = False
-        checked_coins: Set[bytes32] = set()
-        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]
-            removed_dict = await self.wallet_state_manager.trade_manager.get_coins_of_interest()
-            all_coin_names.extend(removed_dict.keys())
-
-            to_check: List[bytes32] = []
-            for coin_name in all_coin_names:
-                if coin_name in checked_coins:
-                    continue
-                else:
-                    to_check.append(coin_name)
-                    checked_coins.add(coin_name)
-                    if len(to_check) == 1000:
-                        break
-
-            msg1 = wallet_protocol.RegisterForCoinUpdates(to_check, uint32(0))
-            new_state: Optional[RespondToCoinUpdates] = await peer.register_interest_in_coin(msg1)
-
-            assert new_state is not None
-            if syncing:
-                # If syncing, completely change over to this peer's information
-                coin_state_before_fork: List[CoinState] = new_state.coin_states
-            else:
-                # Otherwise, we only want to apply changes before the fork point, since we are synced to another peer
-                # We are just validating that there is no missing information
-                coin_state_before_fork = []
-                for coin_state_entry in new_state.coin_states:
-                    if coin_state_entry.spent_height is not None:
-                        if coin_state_entry.spent_height <= fork_height:
-                            coin_state_before_fork.append(coin_state_entry)
-                    elif coin_state_entry.created_height is not None:
-                        if coin_state_entry.created_height <= fork_height:
-                            coin_state_before_fork.append(coin_state_entry)
-
-            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_all_coins = True
-            for coin_name in all_coin_names:
-                if coin_name not in checked_coins:
-                    checked_all_coins = False
-                    break
-
-        end_time = time.time()
-        duration = end_time - start_time
-        self.log.info(f"Sync duration was: {duration}")
+        all_coins: Set[WalletCoinRecord] = await self.wallet_state_manager.coin_store.get_coins_to_check(min_height)
+        all_coin_names: Set[bytes32] = {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.update(removed_dict.keys())
+        all_coin_names.update(await self.wallet_state_manager.interested_store.get_interested_coin_ids())
+        return list(all_coin_names)
 
     async def validate_received_state_from_peer(
         self,
         coin_state: CoinState,
         peer: WSChiaConnection,
         peer_request_cache: PeerRequestCache,
+        fork_height: Optional[uint32],
     ) -> bool:
         """
         Returns all state that is valid and included in the blockchain proved by the weight proof. If return_old_states
         is False, only new states that are not in the coin_store are returned.
         """
         assert self.wallet_state_manager is not None
-        if coin_state.coin.get_hash() in peer_request_cache.states_validated:
+
+        # Only use the cache if we are talking about states before the fork point. If we are evaluating something
+        # in a reorg, we cannot use the cache, since we don't know if it's actually in the new chain after the reorg.
+        if await can_use_peer_request_cache(coin_state, peer_request_cache, fork_height):
             return True
 
         spent_height = coin_state.spent_height
@@ -1101,7 +1039,7 @@ async def validate_received_state_from_peer(
         current = await self.wallet_state_manager.coin_store.get_coin_record(coin_state.coin.name())
         # if remote state is same as current local state we skip validation
 
-        # CoinRecord unspent = height 0, coin state = None. We adjust for comparison bellow
+        # CoinRecord unspent = height 0, coin state = None. We adjust for comparison below
         current_spent_height = None
         if current is not None and current.spent_block_height != 0:
             current_spent_height = current.spent_block_height
@@ -1143,13 +1081,16 @@ async def validate_received_state_from_peer(
         )
 
         if validate_additions_result is False:
+            self.log.warning("Validate false 1")
             await peer.close(9999)
             return False
 
-        # get blocks on top of this block
-        validated = await self.validate_block_inclusion(state_block, peer, peer_request_cache)
-        if not validated:
-            return False
+        # If spent_height is None, we need to validate that the creation block is actually in the longest blockchain.
+        # Otherwise, we don't have to, since we will validate the spent block later.
+        if coin_state.spent_height is None:
+            validated = await self.validate_block_inclusion(state_block, peer, peer_request_cache)
+            if not validated:
+                return False
 
         if spent_height is None and current is not None and current.spent_block_height != 0:
             # Peer is telling us that coin that was previously known to be spent is not spent anymore
@@ -1173,6 +1114,7 @@ async def validate_received_state_from_peer(
                 spent_state_block.foliage_transaction_block.removals_root,
             )
             if validate_removals_result is False:
+                self.log.warning("Validate false 2")
                 await peer.close(9999)
                 return False
             validated = await self.validate_block_inclusion(spent_state_block, peer, peer_request_cache)
@@ -1198,12 +1140,13 @@ async def validate_received_state_from_peer(
                 spent_state_block.foliage_transaction_block.removals_root,
             )
             if validate_removals_result is False:
+                self.log.warning("Validate false 3")
                 await peer.close(9999)
                 return False
             validated = await self.validate_block_inclusion(spent_state_block, peer, peer_request_cache)
             if not validated:
                 return False
-        peer_request_cache.states_validated[coin_state.coin.get_hash()] = coin_state
+        peer_request_cache.states_validated[coin_state.get_hash()] = coin_state
         return True
 
     async def validate_block_inclusion(
@@ -1354,7 +1297,31 @@ async def fetch_puzzle_solution(self, peer, height: uint32, coin: Coin) -> CoinS
             solution_response.response.solution.to_serialized_program(),
         )
 
-    async def fetch_children(self, peer, coin_name) -> List[CoinState]:
+    async def get_coin_state(self, coin_names: List[bytes32], fork_height: Optional[uint32] = None) -> List[CoinState]:
+        assert self.server is not None
+        all_nodes = self.server.connection_by_type[NodeType.FULL_NODE]
+        if len(all_nodes.keys()) == 0:
+            raise ValueError("Not connected to the full node")
+        first_node = list(all_nodes.values())[0]
+        msg = wallet_protocol.RegisterForCoinUpdates(coin_names, uint32(0))
+        coin_state: Optional[RespondToCoinUpdates] = await first_node.register_interest_in_coin(msg)
+        assert coin_state is not None
+
+        if not self.is_trusted(first_node):
+            valid_list = []
+            for coin in coin_state.coin_states:
+                valid = await self.validate_received_state_from_peer(
+                    coin, first_node, self.get_cache_for_peer(first_node), fork_height
+                )
+                if valid:
+                    valid_list.append(coin)
+            return valid_list
+
+        return coin_state.coin_states
+
+    async def fetch_children(
+        self, peer: WSChiaConnection, coin_name: bytes32, fork_height: Optional[uint32] = None
+    ) -> List[CoinState]:
         response: Optional[wallet_protocol.RespondChildren] = await peer.request_children(
             wallet_protocol.RequestChildren(coin_name)
         )
@@ -1365,13 +1332,13 @@ async def fetch_children(self, peer, coin_name) -> List[CoinState]:
             request_cache = self.get_cache_for_peer(peer)
             validated = []
             for state in response.coin_states:
-                valid = await self.validate_received_state_from_peer(state, peer, request_cache)
+                valid = await self.validate_received_state_from_peer(state, peer, request_cache, fork_height)
                 if valid:
                     validated.append(state)
             return validated
         return response.coin_states
 
-    # For RPC only.  You should use wallet_state_manager.add_pending_transaction for normal wallet business.
+    # For RPC only. You should use wallet_state_manager.add_pending_transaction for normal wallet business.
     async def push_tx(self, spend_bundle):
         msg = make_msg(
             ProtocolMessageTypes.send_transaction,
diff --git a/chia/wallet/wallet_node_api.py b/chia/wallet/wallet_node_api.py
index 9e0c6617b142..ec16eef24dc7 100644
--- a/chia/wallet/wallet_node_api.py
+++ b/chia/wallet/wallet_node_api.py
@@ -46,7 +46,8 @@ async def new_peak_wallet(self, peak: wallet_protocol.NewPeakWallet, peer: WSChi
         """
         The full node sent as a new peak
         """
-        await self.wallet_node.new_peak_wallet(peak, peer)
+        self.wallet_node.node_peaks[peer.peer_node_id] = (peak.height, peak.header_hash)
+        await self.wallet_node.new_peak_queue.new_peak_wallet(peak, peer)
 
     @api_request
     async def reject_block_header(self, response: wallet_protocol.RejectHeaderRequest):
@@ -135,10 +136,11 @@ async def respond_header_blocks(self, request: wallet_protocol.RespondHeaderBloc
     async def reject_header_blocks(self, request: wallet_protocol.RejectHeaderBlocks):
         self.log.warning(f"Reject header blocks: {request}")
 
+    @execute_task
     @peer_required
     @api_request
     async def coin_state_update(self, request: wallet_protocol.CoinStateUpdate, peer: WSChiaConnection):
-        await self.wallet_node.state_update_received(request, peer)
+        await self.wallet_node.new_peak_queue.full_node_state_updated(request, peer)
 
     @api_request
     async def respond_to_ph_update(self, request: wallet_protocol.RespondToPhUpdates):
diff --git a/chia/wallet/wallet_puzzle_store.py b/chia/wallet/wallet_puzzle_store.py
index 8034a7237810..c68df2898740 100644
--- a/chia/wallet/wallet_puzzle_store.py
+++ b/chia/wallet/wallet_puzzle_store.py
@@ -308,7 +308,7 @@ async def index_for_puzzle_hash_and_wallet(self, puzzle_hash: bytes32, wallet_id
 
         return None
 
-    async def wallet_info_for_puzzle_hash(self, puzzle_hash: bytes32) -> Optional[Tuple[uint32, WalletType]]:
+    async def wallet_info_for_puzzle_hash(self, puzzle_hash: bytes32) -> Optional[Tuple[int, WalletType]]:
         """
         Returns the derivation path for the puzzle_hash.
         Returns None if not present.
diff --git a/chia/wallet/wallet_state_manager.py b/chia/wallet/wallet_state_manager.py
index d3a060c1d986..654e704e0936 100644
--- a/chia/wallet/wallet_state_manager.py
+++ b/chia/wallet/wallet_state_manager.py
@@ -326,9 +326,12 @@ async def create_more_puzzle_hashes(self, from_zero: bool = False, in_transactio
                         False,
                     )
                 )
-            puzzle_hashes = [record.puzzle_hash for record in derivation_paths]
             await self.puzzle_store.add_derivation_paths(derivation_paths, in_transaction)
-            await self.wallet_node.new_puzzle_hash_created(puzzle_hashes)
+            await self.add_interested_puzzle_hashes(
+                [record.puzzle_hash for record in derivation_paths],
+                [record.wallet_id for record in derivation_paths],
+                in_transaction,
+            )
         if unused > 0:
             await self.puzzle_store.set_used_up_to(uint32(unused - 1), in_transaction)
 
@@ -446,7 +449,7 @@ async def synced(self):
         if latest is None:
             return False
 
-        if latest.height - self.finished_sync_up_to > 2:
+        if latest.height - await self.blockchain.get_finished_sync_up_to() > 1:
             return False
 
         latest_timestamp = self.blockchain.get_latest_timestamp()
@@ -571,13 +574,17 @@ async def unconfirmed_removals_for_wallet(self, wallet_id: int) -> Dict[bytes32,
                 removals[coin.name()] = coin
         return removals
 
-    async def fetch_parent_and_check_for_cat(self, peer, coin_state) -> Tuple[Optional[uint32], Optional[WalletType]]:
+    async def fetch_parent_and_check_for_cat(
+        self, peer: WSChiaConnection, coin_state: CoinState, fork_height: Optional[uint32]
+    ) -> Tuple[Optional[uint32], Optional[WalletType]]:
         if self.is_pool_reward(coin_state.created_height, coin_state.coin.parent_coin_info) or self.is_farmer_reward(
             coin_state.created_height, coin_state.coin.parent_coin_info
         ):
             return None, None
 
-        response: List[CoinState] = await self.wallet_node.get_coin_state([coin_state.coin.parent_coin_info], peer)
+        response: List[CoinState] = await self.wallet_node.get_coin_state(
+            [coin_state.coin.parent_coin_info], fork_height, peer
+        )
         if len(response) == 0:
             self.log.warning(f"Could not find a parent coin with ID: {coin_state.coin.parent_coin_info}")
             return None, None
@@ -624,11 +631,12 @@ async def fetch_parent_and_check_for_cat(self, peer, coin_state) -> Tuple[Option
         return wallet_id, wallet_type
 
     async def new_coin_state(
-        self,
-        coin_states: List[CoinState],
-        peer: WSChiaConnection,
-    ):
-        created_h_none = []
+        self, coin_states: List[CoinState], peer: WSChiaConnection, fork_height: Optional[uint32]
+    ) -> None:
+        # TODO: add comment about what this method does
+
+        # Sort by created height, then add the reorg states (created_height is None) to the end
+        created_h_none: List[CoinState] = []
         for coin_st in coin_states.copy():
             if coin_st.created_height is None:
                 coin_states.remove(coin_st)
@@ -641,10 +649,13 @@ async def new_coin_state(
         trade_coin_removed: List[CoinState] = []
 
         for coin_state_idx, coin_state in enumerate(coin_states):
-            info = await self.get_wallet_id_for_puzzle_hash(coin_state.coin.puzzle_hash)
+            wallet_info: Optional[Tuple[uint32, WalletType]] = 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())
             self.log.debug(f"{coin_state.coin.name()}: {coin_state}")
 
+            # If we already have this coin, and it was spent and confirmed at the same heights, then we return (done)
             if local_record is not None:
                 local_spent = None
                 if local_record.spent_block_height != 0:
@@ -655,15 +666,15 @@ async def new_coin_state(
                 ):
                     continue
 
-            wallet_id = None
-            wallet_type = None
-            if info is not None:
-                wallet_id, wallet_type = info
+            wallet_id: Optional[uint32] = None
+            wallet_type: Optional[WalletType] = None
+            if wallet_info is not None:
+                wallet_id, wallet_type = wallet_info
             elif local_record is not None:
                 wallet_id = uint32(local_record.wallet_id)
                 wallet_type = local_record.wallet_type
             elif coin_state.created_height is not None:
-                wallet_id, wallet_type = await self.fetch_parent_and_check_for_cat(peer, coin_state)
+                wallet_id, wallet_type = await self.fetch_parent_and_check_for_cat(peer, coin_state, fork_height)
 
             if wallet_id is None or wallet_type is None:
                 self.log.info(f"No wallet for coin state: {coin_state}")
@@ -683,6 +694,7 @@ async def new_coin_state(
 
             if coin_state.created_height is None:
                 # TODO implements this coin got reorged
+                # TODO: we need to potentially roll back the pool wallet here
                 pass
             elif coin_state.created_height is not None and coin_state.spent_height is None:
                 await self.coin_added(coin_state.coin, coin_state.created_height, all_txs, wallet_id, wallet_type)
@@ -745,7 +757,7 @@ async def new_coin_state(
                         )
                         await self.tx_store.add_transaction_record(tx_record, False)
 
-                    children = await self.wallet_node.fetch_children(peer, coin_state.coin.name())
+                    children = await self.wallet_node.fetch_children(peer, coin_state.coin.name(), fork_height)
                     assert children is not None
                     additions = [state.coin for state in children]
                     if len(children) > 0:
@@ -820,6 +832,7 @@ async def new_coin_state(
                     if coin_state.spent_height is not None and coin_state.coin.amount == uint64(1):
                         wallet = self.wallets[uint32(record.wallet_id)]
                         curr_coin_state: CoinState = coin_state
+
                         while curr_coin_state.spent_height is not None:
                             cs: CoinSpend = await self.wallet_node.fetch_puzzle_solution(
                                 peer, curr_coin_state.spent_height, curr_coin_state.coin
@@ -839,16 +852,16 @@ async def new_coin_state(
                                 record.wallet_type,
                             )
                             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)
+                            await self.add_interested_coin_ids([new_singleton_coin.name()], True)
                             new_coin_state: List[CoinState] = await self.wallet_node.get_coin_state(
-                                [new_singleton_coin.name()]
+                                [new_singleton_coin.name()], fork_height
                             )
                             assert len(new_coin_state) == 1
                             curr_coin_state = new_coin_state[0]
 
                 # Check if a child is a singleton launcher
                 if children is None:
-                    children = await self.wallet_node.fetch_children(peer, coin_state.coin.name())
+                    children = await self.wallet_node.fetch_children(peer, coin_state.coin.name(), fork_height)
                 assert children is not None
                 for child in children:
                     if child.coin.puzzle_hash != SINGLETON_LAUNCHER_HASH:
@@ -856,18 +869,19 @@ async def new_coin_state(
                     if await self.have_a_pool_wallet_with_launched_id(child.coin.name()):
                         continue
                     if child.spent_height is None:
+                        # TODO handle spending launcher later block
                         continue
                     launcher_spend: Optional[CoinSpend] = await self.wallet_node.fetch_puzzle_solution(
                         peer, coin_state.spent_height, child.coin
                     )
                     if launcher_spend is None:
                         continue
-                    pool_state = None
                     try:
                         pool_state = solution_to_pool_state(launcher_spend)
                     except Exception as e:
                         self.log.debug(f"Not a pool wallet launcher {e}")
                         continue
+
                     # solution_to_pool_state may return None but this may not be an error
                     if pool_state is None:
                         self.log.debug("solution_to_pool_state returned None, ignore and continue")
@@ -887,12 +901,12 @@ async def new_coin_state(
                     await self.coin_added(
                         coin_added, coin_state.spent_height, [], pool_wallet.id(), WalletType(pool_wallet.type())
                     )
-                    await self.interested_store.add_interested_coin_id(coin_added.name(), True)
+                    await self.add_interested_coin_ids([coin_added.name()], True)
 
             else:
                 raise RuntimeError("All cases already handled")  # Logic error, all cases handled
         for coin_state_removed in trade_coin_removed:
-            await self.trade_manager.coins_of_interest_farmed(coin_state_removed)
+            await self.trade_manager.coins_of_interest_farmed(coin_state_removed, fork_height)
 
     async def have_a_pool_wallet_with_launched_id(self, launcher_id: bytes32) -> bool:
         for wallet_id, wallet in self.wallets.items():
@@ -928,7 +942,7 @@ async def get_wallet_id_for_puzzle_hash(self, puzzle_hash: bytes32) -> Optional[
         info = await self.puzzle_store.wallet_info_for_puzzle_hash(puzzle_hash)
         if info is not None:
             wallet_id, wallet_type = info
-            return wallet_id, wallet_type
+            return uint32(wallet_id), wallet_type
 
         interested_wallet_id = await self.interested_store.get_interested_puzzle_hash_wallet_id(puzzle_hash=puzzle_hash)
         if interested_wallet_id is not None:
@@ -936,7 +950,7 @@ async def get_wallet_id_for_puzzle_hash(self, puzzle_hash: bytes32) -> Optional[
             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 uint32(wallet_id), wallet_type
         return None
 
     async def coin_added(
@@ -1052,9 +1066,7 @@ async def add_pending_transaction(self, tx_record: TransactionRecord):
         all_coins_names.extend([coin.name() for coin in tx_record.additions])
         all_coins_names.extend([coin.name() for coin in tx_record.removals])
 
-        nodes = self.server.get_full_node_connections()
-        for node in nodes:
-            await self.wallet_node.subscribe_to_coin_updates(all_coins_names, node)
+        await self.add_interested_coin_ids(all_coins_names)
         self.tx_pending_changed()
         self.state_changed("pending_transaction", tx_record.wallet_id)
 
@@ -1247,16 +1259,19 @@ async def new_peak(self, peak: wallet_protocol.NewPeakWallet):
             if wallet.type() == uint8(WalletType.POOLING_WALLET):
                 await wallet.new_peak(peak.height)
 
-    async def add_interested_puzzle_hash(
-        self, puzzle_hash: bytes32, wallet_id: int, in_transaction: bool = False
+    async def add_interested_puzzle_hashes(
+        self, puzzle_hashes: List[bytes32], wallet_ids: List[int], in_transaction: bool = False
     ) -> None:
-        await self.interested_store.add_interested_puzzle_hash(puzzle_hash, wallet_id, in_transaction)
-        await self.wallet_node.new_puzzle_hash_created([puzzle_hash])
-
-    async def add_interested_coin_id(self, coin_id: bytes32) -> None:
-        nodes = self.server.get_full_node_connections()
-        for node in nodes:
-            await self.wallet_node.subscribe_to_coin_updates([coin_id], node)
+        for puzzle_hash, wallet_id in zip(puzzle_hashes, wallet_ids):
+            await self.interested_store.add_interested_puzzle_hash(puzzle_hash, wallet_id, in_transaction)
+        if len(puzzle_hashes) > 0:
+            await self.wallet_node.new_peak_queue.subscribe_to_puzzle_hashes(puzzle_hashes)
+
+    async def add_interested_coin_ids(self, coin_ids: List[bytes32], in_transaction: bool = False) -> None:
+        for coin_id in coin_ids:
+            await self.interested_store.add_interested_coin_id(coin_id, in_transaction)
+        if len(coin_ids) > 0:
+            await self.wallet_node.new_peak_queue.subscribe_to_coin_ids(coin_ids)
 
     async def delete_trade_transactions(self, trade_id: bytes32):
         txs: List[TransactionRecord] = await self.tx_store.get_transactions_by_trade_id(trade_id)
diff --git a/tests/core/server/test_rate_limits.py b/tests/core/server/test_rate_limits.py
index a1011fbd2172..7833a04ba2a5 100644
--- a/tests/core/server/test_rate_limits.py
+++ b/tests/core/server/test_rate_limits.py
@@ -23,11 +23,11 @@ async def test_too_many_messages(self):
         # Too many messages
         r = RateLimiter(incoming=True)
         new_tx_message = make_msg(ProtocolMessageTypes.new_transaction, bytes([1] * 40))
-        for i in range(3000):
+        for i in range(4900):
             assert r.process_msg_and_check(new_tx_message)
 
         saw_disconnect = False
-        for i in range(3000):
+        for i in range(4900):
             response = r.process_msg_and_check(new_tx_message)
             if not response:
                 saw_disconnect = True
@@ -146,11 +146,11 @@ async def test_periodic_reset(self):
         # Counts reset also
         r = RateLimiter(True, 5)
         new_tx_message = make_msg(ProtocolMessageTypes.new_transaction, bytes([1] * 40))
-        for i in range(3000):
+        for i in range(4900):
             assert r.process_msg_and_check(new_tx_message)
 
         saw_disconnect = False
-        for i in range(3000):
+        for i in range(4900):
             response = r.process_msg_and_check(new_tx_message)
             if not response:
                 saw_disconnect = True
diff --git a/tests/pools/test_pool_rpc.py b/tests/pools/test_pool_rpc.py
index 4e053fa2608d..e9557483e95e 100644
--- a/tests/pools/test_pool_rpc.py
+++ b/tests/pools/test_pool_rpc.py
@@ -23,6 +23,7 @@
 from chia.util.bech32m import encode_puzzle_hash
 from chia.util.byte_types import hexstr_to_bytes
 from chia.wallet.derive_keys import find_authentication_sk, find_owner_sk
+from chia.wallet.wallet_node import WalletNode
 from tests.block_tools import get_plot_dir
 from chia.util.config import load_config
 from chia.util.ints import uint16, uint32
@@ -62,6 +63,14 @@ async def create_pool_plot(p2_singleton_puzzle_hash: bytes32) -> Optional[bytes3
     return plot_id
 
 
+async def wallet_is_synced(wallet_node: WalletNode, full_node_api):
+    assert wallet_node.wallet_state_manager is not None
+    return (
+        await wallet_node.wallet_state_manager.blockchain.get_finished_sync_up_to()
+        == full_node_api.full_node.blockchain.get_peak_height()
+    )
+
+
 @pytest.fixture(scope="session")
 def event_loop():
     loop = asyncio.get_event_loop()
@@ -186,7 +195,7 @@ async def test_create_new_pool_wallet_self_farm(self, one_wallet_node_and_rpc, f
         for summary in summaries_response:
             if WalletType(int(summary["type"])) == WalletType.POOLING_WALLET:
                 assert False
-
+        await time_out_assert(10, wallet_is_synced, True, wallet_node_0, full_node_api)
         creation_tx: TransactionRecord = await client.create_new_pool_wallet(
             our_ph, "", 0, "localhost:5000", "new", "SELF_POOLING", fee
         )
@@ -257,6 +266,8 @@ async def test_create_new_pool_wallet_farm_to_pool(self, one_wallet_node_and_rpc
 
         await time_out_assert(10, wallet_0.get_confirmed_balance, total_block_rewards)
 
+        await time_out_assert(10, wallet_is_synced, True, wallet_node_0, full_node_api)
+
         our_ph = await wallet_0.get_new_puzzlehash()
         summaries_response = await client.get_wallets()
         for summary in summaries_response:
@@ -331,6 +342,7 @@ async def test_create_multiple_pool_wallets(self, one_wallet_node_and_rpc, fee,
         wallet_0 = wallet_node_0.wallet_state_manager.main_wallet
         await time_out_assert(10, wallet_0.get_confirmed_balance, total_block_rewards)
         await time_out_assert(10, wallet_node_0.wallet_state_manager.blockchain.get_peak_height, PREFARMED_BLOCKS)
+        await time_out_assert(10, wallet_is_synced, True, wallet_node_0, full_node_api)
 
         our_ph_1 = await wallet_0.get_new_puzzlehash()
         our_ph_2 = await wallet_0.get_new_puzzlehash()
@@ -390,12 +402,6 @@ async def test_create_multiple_pool_wallets(self, one_wallet_node_and_rpc, fee,
         with pytest.raises(ValueError):
             await client.pw_status(3)
 
-        def wallet_is_synced():
-            return (
-                wallet_node_0.wallet_state_manager.blockchain.get_peak_height()
-                == full_node_api.full_node.blockchain.get_peak_height()
-            )
-
         # Create some CAT wallets to increase wallet IDs
         for i in range(5):
             await asyncio.sleep(2)
@@ -407,7 +413,7 @@ def wallet_is_synced():
             asset_id = bytes.fromhex(res["asset_id"])
             assert len(asset_id) > 0
             await self.farm_blocks(full_node_api, our_ph_2, 6)
-            await time_out_assert(20, wallet_is_synced)
+            await time_out_assert(20, wallet_is_synced, True, wallet_node_0, full_node_api)
             bal_0 = await client.get_wallet_balance(cat_0_id)
             assert bal_0["confirmed_wallet_balance"] == 20
 
@@ -415,6 +421,7 @@ def wallet_is_synced():
         # run this code more than once, since it's slow.
         if fee == 0 and not trusted:
             for i in range(22):
+                await time_out_assert(10, wallet_is_synced, True, wallet_node_0, full_node_api)
                 creation_tx_3: TransactionRecord = await client.create_new_pool_wallet(
                     our_ph_1, "localhost", 5, "localhost:5000", "new", "FARMING_TO_POOL", fee
                 )
@@ -425,7 +432,7 @@ def wallet_is_synced():
                     creation_tx_3.name,
                 )
                 await self.farm_blocks(full_node_api, our_ph_2, 2)
-                await time_out_assert(20, wallet_is_synced)
+                await time_out_assert(20, wallet_is_synced, True, wallet_node_0, full_node_api)
 
                 full_config: Dict = load_config(wallet_0.wallet_state_manager.root_path, "config.yaml")
                 pool_list: List[Dict] = full_config["pool"]["pool_list"]
@@ -447,7 +454,7 @@ def wallet_is_synced():
                             assert owner_sk != auth_sk
 
     @pytest.mark.asyncio
-    @pytest.mark.parametrize("trusted", [True])
+    @pytest.mark.parametrize("trusted", [True, False])
     @pytest.mark.parametrize("fee", [0])
     async def test_absorb_self(self, one_wallet_node_and_rpc, fee, trusted):
         client, wallet_node_0, full_node_api = one_wallet_node_and_rpc
@@ -719,6 +726,7 @@ async def test_self_pooling_to_pooling(self, setup, fee, trusted):
             await time_out_assert(10, wallets[0].get_unconfirmed_balance, total_block_rewards)
             await time_out_assert(10, wallets[0].get_confirmed_balance, total_block_rewards)
             await time_out_assert(10, wallets[0].get_spendable_balance, total_block_rewards)
+            await time_out_assert(10, wallet_is_synced, True, wallet_node_0, full_node_api)
             assert total_block_rewards > 0
 
             summaries_response = await client.get_wallets()
@@ -857,6 +865,7 @@ async def have_chia():
                 return (await wallets[0].get_confirmed_balance()) > 0
 
             await time_out_assert(timeout=WAIT_SECS, function=have_chia)
+            await time_out_assert(10, wallet_is_synced, True, wallet_nodes[0], full_node_api)
 
             creation_tx: TransactionRecord = await client.create_new_pool_wallet(
                 our_ph, "", 0, "localhost:5000", "new", "SELF_POOLING", fee
@@ -872,6 +881,8 @@ async def have_chia():
             await self.farm_blocks(full_node_api, our_ph, 6)
             assert full_node_api.full_node.mempool_manager.get_spendbundle(creation_tx.name) is None
 
+            await time_out_assert(10, wallet_is_synced, True, wallet_nodes[0], full_node_api)
+
             summaries_response = await client.get_wallets()
             wallet_id: Optional[int] = None
             for summary in summaries_response:
@@ -912,6 +923,8 @@ async def status_is_farming_to_pool():
 
             await time_out_assert(timeout=WAIT_SECS, function=status_is_farming_to_pool)
 
+            await time_out_assert(10, wallet_is_synced, True, wallet_nodes[0], full_node_api)
+
             status: PoolWalletInfo = (await client.pw_status(wallet_id))[0]
 
             leave_pool_tx: TransactionRecord = await client.pw_self_pool(wallet_id, fee)
@@ -974,6 +987,7 @@ async def have_chia():
                 return (await wallets[0].get_confirmed_balance()) > 0
 
             await time_out_assert(timeout=WAIT_SECS, function=have_chia)
+            await time_out_assert(10, wallet_is_synced, True, wallet_nodes[0], full_node_api)
 
             creation_tx: TransactionRecord = await client.create_new_pool_wallet(
                 pool_a_ph, "https://pool-a.org", 5, "localhost:5000", "new", "FARMING_TO_POOL", fee
@@ -989,6 +1003,8 @@ async def have_chia():
             await self.farm_blocks(full_node_api, our_ph, 6)
             assert full_node_api.full_node.mempool_manager.get_spendbundle(creation_tx.name) is None
 
+            await time_out_assert(10, wallet_is_synced, True, wallet_nodes[0], full_node_api)
+
             summaries_response = await client.get_wallets()
             wallet_id: Optional[int] = None
             for summary in summaries_response:
@@ -1074,6 +1090,7 @@ async def have_chia():
                 return (await wallets[0].get_confirmed_balance()) > 0
 
             await time_out_assert(timeout=WAIT_SECS, function=have_chia)
+            await time_out_assert(10, wallet_is_synced, True, wallet_nodes[0], full_node_api)
 
             creation_tx: TransactionRecord = await client.create_new_pool_wallet(
                 pool_a_ph, "https://pool-a.org", 5, "localhost:5000", "new", "FARMING_TO_POOL", fee
@@ -1089,6 +1106,8 @@ async def have_chia():
             await self.farm_blocks(full_node_api, our_ph, 6)
             assert full_node_api.full_node.mempool_manager.get_spendbundle(creation_tx.name) is None
 
+            await time_out_assert(5, wallet_is_synced, True, wallet_nodes[0], full_node_api)
+
             summaries_response = await client.get_wallets()
             wallet_id: Optional[int] = None
             for summary in summaries_response:
diff --git a/tests/wallet/test_wallet.py b/tests/wallet/test_wallet.py
index 130905652937..67d76600527b 100644
--- a/tests/wallet/test_wallet.py
+++ b/tests/wallet/test_wallet.py
@@ -167,7 +167,7 @@ async def test_wallet_make_transaction(self, two_wallet_nodes, trusted):
 
     @pytest.mark.parametrize(
         "trusted",
-        [True, False],
+        [False],
     )
     @pytest.mark.asyncio
     async def test_wallet_coinbase_reorg(self, wallet_node, trusted):


Code review comments:
diff_hunk: @@ -0,0 +1,17 @@
+from typing import List
+
+from chia.protocols.wallet_protocol import CoinState
+
+
+def filter_coin_states(all_coins_state: List[CoinState], fork_height: int) -> List[CoinState]:
+    # We only want to apply changes before the fork point, since we are synced to another peer
+    # We are just validating that there is no missing information
+    final_coin_state: List[CoinState] = []
+    for coin_state_entry in all_coins_state:
+        if coin_state_entry.spent_height is not None:
+            if coin_state_entry.spent_height <= fork_height:
+                final_coin_state.append(coin_state_entry)
+        elif coin_state_entry.created_height is not None:
+            if coin_state_entry.created_height <= fork_height:
+                final_coin_state.append(coin_state_entry)
+    return final_coin_state
path: chia/wallet/util/filter_coin_states.py
text: @user1:
If the function is called `filter_foo`, why not use `filter` :^)
I think this is
```python
def accept(coin_state):
    sh = coin_state.spent_height
    if sh is not None:
        return sh <= fork_height
    ch = coin_state.created_height
    return ch is not None and ch <= fork_height

return list(filter(accept, all_coins_state))
```
(And I said this before, but I think half of the list casts in the code base are unnecessary.)