Repository: Chia-Network/chia-blockchain
Pull request: Add a try/except for individual coin states
This looks like a big change but really it's just a big tab to put a large chunk of code inside a try/except.  Currently, if a single state fails during new_coin_state, an exception will be raised to the wallet node which will log an error and then return that it failed to receive the state at all (even though my understanding is that DB changes may have already been made).  This potentially results in not being able to progress past the block with the error causing state in it.  Instead, we should catch exceptions for individual states and just move past them gracefully while logging that we failed to add them.

Problem statement:
Wallet syncing in an infinity loop [[Bug] 
### What happened?

Hi,

My wallet has over 6k NFTs, and suddenly it's not able to get synced. I'm getting always: "ERROR    Error adding states" and my it tries to "perform_atomic_rollback" without success in an infinity loop...

I tried deleting my db and differents PCs with fresh installation, but the issue remain the same on every system.

Now, I'm not able to mint new NFTs or at least transfer my did because never get synced.

### Version

1.5.1

### What platform are you using?

Linux

### What ui mode are you using?

CLI

### Relevant log output

```shell
2022-08-31T18:18:39.971 wallet chia.wallet.wallet_state_manager: INFO     Setting tx_id: 02a9c0cf3d2d97532f8f48029d73a819c0dcee8b085fc1904f21cbbb95b0a7ba to confirmed
2022-08-31T18:18:39.988 wallet DID did:chia:13q0lkt30jysru9078xplncw2x8xm8m3kjsxgk2fdt3w7fa05aymqzqn86a: INFO     DID wallet has been notified that coin was added: b'\x0e-5_\x15f\x9aq\xb47\xdb\x8dpoA\xa2\x9f\x15\xfc\xc5Nu/\xb8\xb5\x85\xccL"&o\xdd':Coin { parent_coin_info: 4ffd9924bec8d58b2236e48482253bc2ac35952cfb364883a27e33bc13e3d024, puzzle_hash: 913064fff26074f8400b70c1fb8a22bdb99c20709e8874b64679d55734b76e74, amount: 1000000 }
2022-08-31T18:18:40.493 wallet DID did:chia:13q0lkt30jysru9078xplncw2x8xm8m3kjsxgk2fdt3w7fa05aymqzqn86a: INFO     Adding parent b'\x0e-5_\x15f\x9aq\xb47\xdb\x8dpoA\xa2\x9f\x15\xfc\xc5Nu/\xb8\xb5\x85\xccL"&o\xdd': {'amount': 1000000,
 'inner_puzzle_hash': '0x5b6d7e41c46099b3bff4d0bfa40f385ff46653f6f7692be7890baf6ca7128db2',
 'parent_name': '0x4ffd9924bec8d58b2236e48482253bc2ac35952cfb364883a27e33bc13e3d024'}
2022-08-31T18:18:41.002 wallet chia.wallet.wallet_node    : ERROR    Error adding states..  Traceback (most recent call last):
  File "chia/wallet/wallet_node.py", line 791, in receive_state_from_peer
  File "chia/wallet/wallet_state_manager.py", line 934, in new_coin_state
  File "chia/wallet/wallet_state_manager.py", line 1330, in coin_added
  File "chia/wallet/did_wallet/did_wallet.py", line 386, in coin_added
AssertionError

2022-08-31T18:18:42.223 wallet chia.wallet.wallet_blockchain: INFO     Peak set to: 2479065 timestamp: 1661983771
2022-08-31T18:18:42.225 wallet chia.wallet.wallet_node    : INFO     Starting sync trusted: True to peer 192.168.21.123
2022-08-31T18:18:42.226 wallet chia.wallet.wallet_node    : INFO     perform_atomic_rollback to 2467068
```

Hints from issue discussion:
Can you provide a log (debug.log file) and sent it to me privately on keybase (sorgente711) or to trepca. @trepca 
you might be hitting some rate limits on `request_puzzle_solution` locally and timing out..
> Can you provide a log (debug.log file) and sent it to me privately on keybase (sorgente711) or to trepca. @trepca

You can check your keybase....
@gonner22 thanks, while we're investigating this further, can you try the latest dev build here: https://www.chia.net/download/ (scroll down to "Latest Dev" header) 
we've added a lot of improvements to the wallet to handle larger volume of NFTs 
> @gonner22 thanks, while we're investigating this further, can you try the latest dev build here: https://www.chia.net/download/ (scroll down to "Latest Dev" header) we've added a lot of improvements to the wallet to handle larger volume of NFTs

Thanks for your reply. Is there a way to update my CLI using this method?

cd chia-blockchain
. ./activate
chia stop -d all
deactivate
git fetch
git checkout latest
git reset --hard FETCH_HEAD --recurse-submodules
git status
sh install.sh

Is there a development branch??
Yes, development branch is `main`. Replace `git checkout latest` with `git checkout main`.
I tested the latest dev header, as you can see here:

git log -1
commit 8db1943cde59fc99fe9dd2a2266b652a9efb4bbc (HEAD -> main, origin/main, origin/HEAD)
Author: Kyle Altendorf <sda@fstab.net>
Date:   Wed Sep 7 23:46:20 2022 -0400

    fix datalayer upsert test flakes (#13353) (#13372)
    
    (cherry picked from commit 9b9868d3ef1f015d13b2a3efd48ec8f64b3441b7)

I have the same issue, my wallet continues in an infinity loop of syncing, but now a new error is displayed on the logs: 
 wallet chia.wallet.nft_wallet.nft_off_chain: ERROR    Cannot get off-chain metadata of ...

If you have also a keybase user, I can send you the debug.log
For those having issues with the wallet 1.5.1 stuck with thousands NFTs, Chia just solved the issue. The new commit is now here quex.new_coin_state_exceptions and you can use it following this upgrade procedure:

cd chia-blockchain
. ./activate
chia stop -d all
deactivate
git fetch
git checkout quex.new_coin_state_exceptions
git reset --hard FETCH_HEAD --recurse-submodules
git status
sh install.sh

It will be available soon on the development branch and for the latest branch will take some time


thanks @mariano54 and @trepca

Commit message: tiny bad merge

Patch to review:
diff --git a/chia/wallet/wallet_node.py b/chia/wallet/wallet_node.py
index 968d2e328360..37cdb5670f15 100644
--- a/chia/wallet/wallet_node.py
+++ b/chia/wallet/wallet_node.py
@@ -130,6 +130,7 @@ class WalletNode:
 
     _shut_down: bool = False
     _process_new_subscriptions_task: Optional[asyncio.Task] = None
+    _retry_failed_states_task: Optional[asyncio.Task] = None
     _primary_peer_sync_task: Optional[asyncio.Task] = None
     _secondary_peer_sync_task: Optional[asyncio.Task] = None
 
@@ -276,10 +277,12 @@ async def _start(
             self.wallet_state_manager.set_callback(self.state_changed_callback)
 
         self.last_wallet_tx_resend_time = int(time.time())
+        self.last_state_retry_time = int(time.time())
         self.wallet_tx_resend_timeout_secs = self.config.get("tx_resend_timeout_secs", 60 * 60)
         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._retry_failed_states_task = asyncio.create_task(self._retry_failed_states())
 
         self.sync_event = asyncio.Event()
         self.log_in(private_key)
@@ -299,6 +302,8 @@ def _close(self):
 
         if self._process_new_subscriptions_task is not None:
             self._process_new_subscriptions_task.cancel()
+        if self._retry_failed_states_task is not None:
+            self._retry_failed_states_task.cancel()
         if self._primary_peer_sync_task is not None:
             self._primary_peer_sync_task.cancel()
         if self._secondary_peer_sync_task is not None:
@@ -377,6 +382,42 @@ async def _messages_to_resend(self) -> List[Tuple[Message, Set[bytes32]]]:
 
         return messages
 
+    async def _retry_failed_states(self):
+        while not self._shut_down:
+            try:
+                await asyncio.sleep(5)
+                current_time = time.time()
+                if self.last_state_retry_time < current_time - 10:
+                    self.last_state_retry_time = current_time
+                    if self.wallet_state_manager is None:
+                        continue
+                    states_to_retry = await self.wallet_state_manager.retry_store.get_all_states_to_retry()
+                    for state, peer_id, fork_height in states_to_retry:
+                        matching_peer = tuple(
+                            p for p in self.server.get_full_node_connections() if p.peer_node_id == peer_id
+                        )
+                        if len(matching_peer) == 0:
+                            peer = self.get_full_node_peer()
+                            self.log.info(
+                                f"disconnected from peer {peer_id}, state will retry with {peer.peer_node_id}"
+                            )
+                        else:
+                            peer = matching_peer[0]
+                        async with self.wallet_state_manager.db_wrapper.writer():
+                            self.log.info(f"retrying coin_state: {state}")
+                            try:
+                                await self.wallet_state_manager.new_coin_state(
+                                    [state], peer, None if fork_height == 0 else fork_height, False
+                                )
+                            except Exception as e:
+                                tb = traceback.format_exc()
+                                self.log.error(f"Exception while adding states.. : {e} {tb}")
+                            else:
+                                await self.wallet_state_manager.blockchain.clean_block_records()
+            except asyncio.CancelledError:
+                self.log.info("Retry task cancelled, exiting.")
+                raise
+
     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
@@ -738,7 +779,7 @@ async def receive_and_validate(inner_states: List[CoinState], inner_idx_start: i
                                 f"{inner_idx_start + len(inner_states) - 1}/ {len(items)})"
                             )
                             try:
-                                await self.wallet_state_manager.new_coin_state(valid_states, peer, fork_height)
+                                await self.wallet_state_manager.new_coin_state(valid_states, peer, fork_height, False)
 
                                 if update_finished_height:
                                     if len(cs_heights) == 1:
@@ -777,7 +818,7 @@ async def receive_and_validate(inner_states: List[CoinState], inner_idx_start: i
                 async with self.wallet_state_manager.db_wrapper.writer():
                     try:
                         self.log.info(f"new coin state received ({idx}-" f"{idx + len(states) - 1}/ {len(items)})")
-                        await self.wallet_state_manager.new_coin_state(states, peer, fork_height)
+                        await self.wallet_state_manager.new_coin_state(states, peer, fork_height, True)
                         if update_finished_height:
                             await self.wallet_state_manager.blockchain.set_finished_sync_up_to(
                                 last_change_height_cs(states[-1]) - 1
diff --git a/chia/wallet/wallet_retry_store.py b/chia/wallet/wallet_retry_store.py
new file mode 100644
index 000000000000..d6426de8f499
--- /dev/null
+++ b/chia/wallet/wallet_retry_store.py
@@ -0,0 +1,52 @@
+from typing import List, Optional, Tuple
+
+from chia_rs import CoinState
+
+from chia.types.blockchain_format.sized_bytes import bytes32
+from chia.util.db_wrapper import DBWrapper2
+from chia.util.ints import uint32
+
+
+class WalletRetryStore:
+    """
+    Persistent coin states that we have received but failed to add
+    """
+
+    db_wrapper: DBWrapper2
+
+    @classmethod
+    async def create(cls, db_wrapper: DBWrapper2) -> "WalletRetryStore":
+        self = cls()
+        self.db_wrapper = db_wrapper
+        async with self.db_wrapper.writer_maybe_transaction() as conn:
+            await conn.execute(
+                "CREATE TABLE IF NOT EXISTS retry_store(" " coin_state blob," " peer blob," " fork_height int)"
+            )
+
+        return self
+
+    async def get_all_states_to_retry(self) -> List[Tuple[CoinState, bytes32, uint32]]:
+        """
+        Return all states that were failed to sync
+        """
+
+        async with self.db_wrapper.reader_no_transaction() as conn:
+            rows = await conn.execute_fetchall("SELECT * from retry_store")
+
+        return [(CoinState.from_bytes(row[0]), bytes32(row[1]), uint32(row[2])) for row in rows]
+
+    async def add_state(self, state: CoinState, peer_id: bytes32, fork_height: Optional[uint32]) -> None:
+        """
+        Adds object to key val store. Obj MUST support __bytes__ and bytes() methods.
+        """
+        async with self.db_wrapper.writer_maybe_transaction() as conn:
+            cursor = await conn.execute(
+                "INSERT OR IGNORE INTO retry_store VALUES(?, ?, ?)",
+                (bytes(state), peer_id, 0 if fork_height is None else fork_height),
+            )
+            await cursor.close()
+
+    async def remove_state(self, state: CoinState) -> None:
+        async with self.db_wrapper.writer_maybe_transaction() as conn:
+            cursor = await conn.execute("DELETE FROM retry_store where coin_state=?", (bytes(state),))
+            await cursor.close()
diff --git a/chia/wallet/wallet_state_manager.py b/chia/wallet/wallet_state_manager.py
index 15a47b9af2fe..53789049e4ea 100644
--- a/chia/wallet/wallet_state_manager.py
+++ b/chia/wallet/wallet_state_manager.py
@@ -4,6 +4,7 @@
 import multiprocessing
 import multiprocessing.context
 import time
+import traceback
 from datetime import datetime
 from collections import defaultdict
 from pathlib import Path
@@ -77,6 +78,7 @@
 from chia.wallet.wallet_nft_store import WalletNftStore
 from chia.wallet.wallet_pool_store import WalletPoolStore
 from chia.wallet.wallet_puzzle_store import WalletPuzzleStore
+from chia.wallet.wallet_retry_store import WalletRetryStore
 from chia.wallet.wallet_transaction_store import WalletTransactionStore
 from chia.wallet.wallet_user_store import WalletUserStore
 from chia.wallet.uncurried_puzzle import uncurry_puzzle
@@ -119,6 +121,7 @@ class WalletStateManager:
     blockchain: WalletBlockchain
     coin_store: WalletCoinStore
     interested_store: WalletInterestedStore
+    retry_store: WalletRetryStore
     multiprocessing_context: multiprocessing.context.BaseContext
     server: ChiaServer
     root_path: Path
@@ -194,6 +197,7 @@ def sql_trace_callback(req: str) -> None:
         self.pool_store = await WalletPoolStore.create(self.db_wrapper)
         self.dl_store = await DataLayerStore.create(self.db_wrapper)
         self.interested_store = await WalletInterestedStore.create(self.db_wrapper)
+        self.retry_store = await WalletRetryStore.create(self.db_wrapper)
         self.default_cats = DEFAULT_CATS
 
         self.wallet_node = wallet_node
@@ -910,6 +914,7 @@ async def new_coin_state(
         coin_states: List[CoinState],
         peer: WSChiaConnection,
         fork_height: Optional[uint32],
+        retry: bool = False,
     ) -> None:
         # TODO: add comment about what this method does
         # Input states should already be sorted by cs_height, with reorgs at the beginning
@@ -932,331 +937,355 @@ async def new_coin_state(
 
         assert len(local_records) == len(coin_states)
         for coin_state, local_record in zip(coin_states, local_records):
-            existing: Optional[WalletCoinRecord]
-            coin_name: bytes32 = coin_state.coin.name()
-            wallet_info: Optional[Tuple[uint32, WalletType]] = await self.get_wallet_id_for_puzzle_hash(
-                coin_state.coin.puzzle_hash
-            )
-            self.log.debug("%s: %s", 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:
-                    local_spent = local_record.spent_block_height
-                if (
-                    local_spent == coin_state.spent_height
-                    and local_record.confirmed_block_height == coin_state.created_height
-                ):
-                    continue
-
-            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.determine_coin_type(peer, coin_state, fork_height)
-                potential_dl = self.get_dl_wallet()
-                if potential_dl is not None:
-                    if await potential_dl.get_singleton_record(coin_state.coin.name()) is not None:
-                        wallet_id = potential_dl.id()
-                        wallet_type = WalletType(potential_dl.type())
-
-            if wallet_id is None or wallet_type is None:
-                self.log.debug(f"No wallet for coin state: {coin_state}")
-                continue
-
-            # Update the DB to signal that we used puzzle hashes up to this one
-            derivation_index = ph_to_index_cache.get(coin_state.coin.puzzle_hash)
-            if derivation_index is None:
-                derivation_index = await self.puzzle_store.index_for_puzzle_hash(coin_state.coin.puzzle_hash)
-            if derivation_index is not None:
-                ph_to_index_cache.put(coin_state.coin.puzzle_hash, derivation_index)
-                if derivation_index > used_up_to:
-                    await self.puzzle_store.set_used_up_to(derivation_index)
-                    used_up_to = max(used_up_to, derivation_index)
-
-            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
-            # if the new coin has not been spent (i.e not ephemeral)
-            elif coin_state.created_height is not None and coin_state.spent_height is None:
-                if local_record is None:
-                    await self.coin_added(
-                        coin_state.coin,
-                        uint32(coin_state.created_height),
-                        all_unconfirmed,
-                        wallet_id,
-                        wallet_type,
-                        peer,
-                        coin_name,
+            try:
+                async with self.db_wrapper.writer():
+                    existing: Optional[WalletCoinRecord]
+                    coin_name: bytes32 = coin_state.coin.name()
+                    wallet_info: Optional[Tuple[uint32, WalletType]] = await self.get_wallet_id_for_puzzle_hash(
+                        coin_state.coin.puzzle_hash
                     )
+                    self.log.debug("%s: %s", coin_name, coin_state)
 
-            # if the coin has been spent
-            elif coin_state.created_height is not None and coin_state.spent_height is not None:
-                self.log.debug("Coin Removed: %s", coin_state)
-                if coin_name in trade_removals:
-                    trade_coin_removed.append(coin_state)
-                children: Optional[List[CoinState]] = None
-                record = local_record
-                if record is None:
-                    farmer_reward = False
-                    pool_reward = False
-                    tx_type: int
-                    if self.is_farmer_reward(uint32(coin_state.created_height), coin_state.coin):
-                        farmer_reward = True
-                        tx_type = TransactionType.FEE_REWARD.value
-                    elif self.is_pool_reward(uint32(coin_state.created_height), coin_state.coin):
-                        pool_reward = True
-                        tx_type = TransactionType.COINBASE_REWARD.value
-                    else:
-                        tx_type = TransactionType.INCOMING_TX.value
-                    record = WalletCoinRecord(
-                        coin_state.coin,
-                        uint32(coin_state.created_height),
-                        uint32(coin_state.spent_height),
-                        True,
-                        farmer_reward or pool_reward,
-                        wallet_type,
-                        wallet_id,
-                    )
-                    await self.coin_store.add_coin_record(record)
-                    # Coin first received
-                    parent_coin_record: Optional[WalletCoinRecord] = await self.coin_store.get_coin_record(
-                        coin_state.coin.parent_coin_info
-                    )
-                    if parent_coin_record is not None and wallet_type.value == parent_coin_record.wallet_type:
-                        change = True
-                    else:
-                        change = False
-
-                    if not change:
-                        created_timestamp = await self.wallet_node.get_timestamp_for_height(coin_state.created_height)
-                        tx_record = TransactionRecord(
-                            confirmed_at_height=uint32(coin_state.created_height),
-                            created_at_time=uint64(created_timestamp),
-                            to_puzzle_hash=(await self.convert_puzzle_hash(wallet_id, coin_state.coin.puzzle_hash)),
-                            amount=uint64(coin_state.coin.amount),
-                            fee_amount=uint64(0),
-                            confirmed=True,
-                            sent=uint32(0),
-                            spend_bundle=None,
-                            additions=[coin_state.coin],
-                            removals=[],
-                            wallet_id=wallet_id,
-                            sent_to=[],
-                            trade_id=None,
-                            type=uint32(tx_type),
-                            name=bytes32(token_bytes()),
-                            memos=[],
-                        )
-                        await self.tx_store.add_transaction_record(tx_record)
-
-                    children = await self.wallet_node.fetch_children(coin_name, peer=peer, fork_height=fork_height)
-                    assert children is not None
-                    additions = [state.coin for state in children]
-                    if len(children) > 0:
-                        fee = 0
-
-                        to_puzzle_hash = None
-                        # Find coin that doesn't belong to us
-                        amount = 0
-                        for coin in additions:
-                            derivation_record = await self.puzzle_store.get_derivation_record_for_puzzle_hash(
-                                coin.puzzle_hash
-                            )
-                            if derivation_record is None:
-                                to_puzzle_hash = coin.puzzle_hash
-                                amount += coin.amount
+                    # If we already have this coin, & it was spent & confirmed at the same heights, then return (done)
+                    if local_record is not None:
+                        local_spent = None
+                        if local_record.spent_block_height != 0:
+                            local_spent = local_record.spent_block_height
+                        if (
+                            local_spent == coin_state.spent_height
+                            and local_record.confirmed_block_height == coin_state.created_height
+                        ):
+                            continue
+
+                    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.determine_coin_type(peer, coin_state, fork_height)
+                        potential_dl = self.get_dl_wallet()
+                        if potential_dl is not None:
+                            if await potential_dl.get_singleton_record(coin_state.coin.name()) is not None:
+                                wallet_id = potential_dl.id()
+                                wallet_type = WalletType(potential_dl.type())
+
+                    if wallet_id is None or wallet_type is None:
+                        self.log.debug(f"No wallet for coin state: {coin_state}")
+                        continue
 
-                        if to_puzzle_hash is None:
-                            to_puzzle_hash = additions[0].puzzle_hash
+                    # Update the DB to signal that we used puzzle hashes up to this one
+                    derivation_index = ph_to_index_cache.get(coin_state.coin.puzzle_hash)
+                    if derivation_index is None:
+                        derivation_index = await self.puzzle_store.index_for_puzzle_hash(coin_state.coin.puzzle_hash)
+                    if derivation_index is not None:
+                        ph_to_index_cache.put(coin_state.coin.puzzle_hash, derivation_index)
+                        if derivation_index > used_up_to:
+                            await self.puzzle_store.set_used_up_to(derivation_index)
+                            used_up_to = max(used_up_to, derivation_index)
+
+                    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
+                    # if the new coin has not been spent (i.e not ephemeral)
+                    elif coin_state.created_height is not None and coin_state.spent_height is None:
+                        if local_record is None:
+                            await self.coin_added(
+                                coin_state.coin,
+                                uint32(coin_state.created_height),
+                                all_unconfirmed,
+                                wallet_id,
+                                wallet_type,
+                                peer,
+                                coin_name,
+                            )
 
-                        spent_timestamp = await self.wallet_node.get_timestamp_for_height(coin_state.spent_height)
+                    # if the coin has been spent
+                    elif coin_state.created_height is not None and coin_state.spent_height is not None:
+                        self.log.debug("Coin Removed: %s", coin_state)
+                        if coin_name in trade_removals:
+                            trade_coin_removed.append(coin_state)
+                        children: Optional[List[CoinState]] = None
+                        record = local_record
+                        if record is None:
+                            farmer_reward = False
+                            pool_reward = False
+                            tx_type: int
+                            if self.is_farmer_reward(uint32(coin_state.created_height), coin_state.coin):
+                                farmer_reward = True
+                                tx_type = TransactionType.FEE_REWARD.value
+                            elif self.is_pool_reward(uint32(coin_state.created_height), coin_state.coin):
+                                pool_reward = True
+                                tx_type = TransactionType.COINBASE_REWARD.value
+                            else:
+                                tx_type = TransactionType.INCOMING_TX.value
+                            record = WalletCoinRecord(
+                                coin_state.coin,
+                                uint32(coin_state.created_height),
+                                uint32(coin_state.spent_height),
+                                True,
+                                farmer_reward or pool_reward,
+                                wallet_type,
+                                wallet_id,
+                            )
+                            await self.coin_store.add_coin_record(record)
+                            # Coin first received
+                            parent_coin_record: Optional[WalletCoinRecord] = await self.coin_store.get_coin_record(
+                                coin_state.coin.parent_coin_info
+                            )
+                            if parent_coin_record is not None and wallet_type.value == parent_coin_record.wallet_type:
+                                change = True
+                            else:
+                                change = False
+
+                            if not change:
+                                created_timestamp = await self.wallet_node.get_timestamp_for_height(
+                                    coin_state.created_height
+                                )
+                                tx_record = TransactionRecord(
+                                    confirmed_at_height=uint32(coin_state.created_height),
+                                    created_at_time=uint64(created_timestamp),
+                                    to_puzzle_hash=(
+                                        await self.convert_puzzle_hash(wallet_id, coin_state.coin.puzzle_hash)
+                                    ),
+                                    amount=uint64(coin_state.coin.amount),
+                                    fee_amount=uint64(0),
+                                    confirmed=True,
+                                    sent=uint32(0),
+                                    spend_bundle=None,
+                                    additions=[coin_state.coin],
+                                    removals=[],
+                                    wallet_id=wallet_id,
+                                    sent_to=[],
+                                    trade_id=None,
+                                    type=uint32(tx_type),
+                                    name=bytes32(token_bytes()),
+                                    memos=[],
+                                )
+                                await self.tx_store.add_transaction_record(tx_record)
 
-                        # Reorg rollback adds reorged transactions so it's possible there is tx_record already
-                        # Even though we are just adding coin record to the db (after reorg)
-                        tx_records: List[TransactionRecord] = []
-                        for out_tx_record in all_unconfirmed:
-                            for rem_coin in out_tx_record.removals:
-                                if rem_coin == coin_state.coin:
-                                    tx_records.append(out_tx_record)
+                            children = await self.wallet_node.fetch_children(
+                                coin_name, peer=peer, fork_height=fork_height
+                            )
+                            assert children is not None
+                            additions = [state.coin for state in children]
+                            if len(children) > 0:
+                                fee = 0
+
+                                to_puzzle_hash = None
+                                # Find coin that doesn't belong to us
+                                amount = 0
+                                for coin in additions:
+                                    derivation_record = await self.puzzle_store.get_derivation_record_for_puzzle_hash(
+                                        coin.puzzle_hash
+                                    )
+                                    if derivation_record is None:
+                                        to_puzzle_hash = coin.puzzle_hash
+                                        amount += coin.amount
+
+                                if to_puzzle_hash is None:
+                                    to_puzzle_hash = additions[0].puzzle_hash
+
+                                spent_timestamp = await self.wallet_node.get_timestamp_for_height(
+                                    coin_state.spent_height
+                                )
 
-                        if len(tx_records) > 0:
-                            for tx_record in tx_records:
-                                await self.tx_store.set_confirmed(tx_record.name, uint32(coin_state.spent_height))
+                                # Reorg rollback adds reorged transactions so it's possible there is tx_record already
+                                # Even though we are just adding coin record to the db (after reorg)
+                                tx_records: List[TransactionRecord] = []
+                                for out_tx_record in all_unconfirmed:
+                                    for rem_coin in out_tx_record.removals:
+                                        if rem_coin == coin_state.coin:
+                                            tx_records.append(out_tx_record)
+
+                                if len(tx_records) > 0:
+                                    for tx_record in tx_records:
+                                        await self.tx_store.set_confirmed(
+                                            tx_record.name, uint32(coin_state.spent_height)
+                                        )
+                                else:
+                                    tx_record = TransactionRecord(
+                                        confirmed_at_height=uint32(coin_state.spent_height),
+                                        created_at_time=uint64(spent_timestamp),
+                                        to_puzzle_hash=(await self.convert_puzzle_hash(wallet_id, to_puzzle_hash)),
+                                        amount=uint64(int(amount)),
+                                        fee_amount=uint64(fee),
+                                        confirmed=True,
+                                        sent=uint32(0),
+                                        spend_bundle=None,
+                                        additions=additions,
+                                        removals=[coin_state.coin],
+                                        wallet_id=wallet_id,
+                                        sent_to=[],
+                                        trade_id=None,
+                                        type=uint32(TransactionType.OUTGOING_TX.value),
+                                        name=bytes32(token_bytes()),
+                                        memos=[],
+                                    )
+
+                                    await self.tx_store.add_transaction_record(tx_record)
                         else:
-                            tx_record = TransactionRecord(
-                                confirmed_at_height=uint32(coin_state.spent_height),
-                                created_at_time=uint64(spent_timestamp),
-                                to_puzzle_hash=(await self.convert_puzzle_hash(wallet_id, to_puzzle_hash)),
-                                amount=uint64(int(amount)),
-                                fee_amount=uint64(fee),
-                                confirmed=True,
-                                sent=uint32(0),
-                                spend_bundle=None,
-                                additions=additions,
-                                removals=[coin_state.coin],
-                                wallet_id=wallet_id,
-                                sent_to=[],
-                                trade_id=None,
-                                type=uint32(TransactionType.OUTGOING_TX.value),
-                                name=bytes32(token_bytes()),
-                                memos=[],
+                            await self.coin_store.set_spent(coin_name, uint32(coin_state.spent_height))
+                            rem_tx_records: List[TransactionRecord] = []
+                            for out_tx_record in all_unconfirmed:
+                                for rem_coin in out_tx_record.removals:
+                                    if rem_coin == coin_state.coin:
+                                        rem_tx_records.append(out_tx_record)
+
+                            for tx_record in rem_tx_records:
+                                await self.tx_store.set_confirmed(tx_record.name, uint32(coin_state.spent_height))
+                        for unconfirmed_record in all_unconfirmed:
+                            for rem_coin in unconfirmed_record.removals:
+                                if rem_coin == coin_state.coin:
+                                    self.log.info(f"Setting tx_id: {unconfirmed_record.name} to confirmed")
+                                    await self.tx_store.set_confirmed(
+                                        unconfirmed_record.name, uint32(coin_state.spent_height)
+                                    )
+
+                        if record.wallet_type == WalletType.POOLING_WALLET:
+                            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(
+                                        curr_coin_state.spent_height, curr_coin_state.coin, peer
+                                    )
+                                    success = await wallet.apply_state_transition(cs, curr_coin_state.spent_height)
+                                    if not success:
+                                        break
+                                    new_singleton_coin: Optional[Coin] = wallet.get_next_interesting_coin(cs)
+                                    if new_singleton_coin is None:
+                                        # No more singleton (maybe destroyed?)
+                                        break
+
+                                    coin_name = new_singleton_coin.name()
+                                    existing = await self.coin_store.get_coin_record(coin_name)
+                                    if existing is None:
+                                        await self.coin_added(
+                                            new_singleton_coin,
+                                            uint32(coin_state.spent_height),
+                                            [],
+                                            uint32(record.wallet_id),
+                                            record.wallet_type,
+                                            peer,
+                                            coin_name,
+                                        )
+                                    await self.coin_store.set_spent(
+                                        curr_coin_state.coin.name(), uint32(curr_coin_state.spent_height)
+                                    )
+                                    await self.add_interested_coin_ids([new_singleton_coin.name()])
+                                    new_coin_state: List[CoinState] = await self.wallet_node.get_coin_state(
+                                        [coin_name], peer=peer, fork_height=fork_height
+                                    )
+                                    assert len(new_coin_state) == 1
+                                    curr_coin_state = new_coin_state[0]
+                        if record.wallet_type == WalletType.DATA_LAYER:
+                            singleton_spend = await self.wallet_node.fetch_puzzle_solution(
+                                coin_state.spent_height, coin_state.coin, peer
+                            )
+                            dl_wallet = self.wallets[uint32(record.wallet_id)]
+                            await dl_wallet.singleton_removed(
+                                singleton_spend,
+                                coin_state.spent_height,
                             )
 
-                            await self.tx_store.add_transaction_record(tx_record)
-                else:
-                    await self.coin_store.set_spent(coin_name, uint32(coin_state.spent_height))
-                    rem_tx_records: List[TransactionRecord] = []
-                    for out_tx_record in all_unconfirmed:
-                        for rem_coin in out_tx_record.removals:
-                            if rem_coin == coin_state.coin:
-                                rem_tx_records.append(out_tx_record)
-
-                    for tx_record in rem_tx_records:
-                        await self.tx_store.set_confirmed(tx_record.name, uint32(coin_state.spent_height))
-                for unconfirmed_record in all_unconfirmed:
-                    for rem_coin in unconfirmed_record.removals:
-                        if rem_coin == coin_state.coin:
-                            self.log.info(f"Setting tx_id: {unconfirmed_record.name} to confirmed")
-                            await self.tx_store.set_confirmed(unconfirmed_record.name, uint32(coin_state.spent_height))
-
-                if record.wallet_type == WalletType.POOLING_WALLET:
-                    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(
-                                curr_coin_state.spent_height, curr_coin_state.coin, peer
+                        elif record.wallet_type == WalletType.NFT:
+                            if coin_state.spent_height is not None:
+                                nft_wallet = self.wallets[uint32(record.wallet_id)]
+                                await nft_wallet.remove_coin(coin_state.coin, coin_state.spent_height)
+
+                        # Check if a child is a singleton launcher
+                        if children is None:
+                            children = await self.wallet_node.fetch_children(
+                                coin_name, peer=peer, fork_height=fork_height
+                            )
+                        assert children is not None
+                        for child in children:
+                            if child.coin.puzzle_hash != SINGLETON_LAUNCHER_HASH:
+                                continue
+                            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(
+                                coin_state.spent_height, child.coin, peer
+                            )
+                            if launcher_spend is None:
+                                continue
+                            try:
+                                pool_state = solution_to_pool_state(launcher_spend)
+                                assert pool_state is not None
+                            except (AssertionError, ValueError) as e:
+                                self.log.debug(f"Not a pool wallet launcher {e}")
+                                matched, inner_puzhash = await DataLayerWallet.match_dl_launcher(launcher_spend)
+                                if (
+                                    matched
+                                    and inner_puzhash is not None
+                                    and (await self.puzzle_store.puzzle_hash_exists(inner_puzhash))
+                                ):
+                                    for _, wallet in self.wallets.items():
+                                        if wallet.type() == WalletType.DATA_LAYER.value:
+                                            dl_wallet = wallet
+                                            break
+                                    else:  # No DL wallet exists yet
+                                        dl_wallet = await DataLayerWallet.create_new_dl_wallet(
+                                            self,
+                                            self.main_wallet,
+                                        )
+                                    await dl_wallet.track_new_launcher_id(
+                                        child.coin.name(),
+                                        peer,
+                                        spend=launcher_spend,
+                                        height=child.spent_height,
+                                    )
+                                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")
+                                continue
+
+                            assert child.spent_height is not None
+                            pool_wallet = await PoolWallet.create(
+                                self,
+                                self.main_wallet,
+                                child.coin.name(),
+                                [launcher_spend],
+                                uint32(child.spent_height),
+                                name="pool_wallet",
                             )
-                            success = await wallet.apply_state_transition(cs, curr_coin_state.spent_height)
-                            if not success:
-                                break
-                            new_singleton_coin: Optional[Coin] = wallet.get_next_interesting_coin(cs)
-                            if new_singleton_coin is None:
-                                # No more singleton (maybe destroyed?)
-                                break
-
-                            coin_name = new_singleton_coin.name()
+                            launcher_spend_additions = launcher_spend.additions()
+                            assert len(launcher_spend_additions) == 1
+                            coin_added = launcher_spend_additions[0]
+                            coin_name = coin_added.name()
                             existing = await self.coin_store.get_coin_record(coin_name)
                             if existing is None:
                                 await self.coin_added(
-                                    new_singleton_coin,
+                                    coin_added,
                                     uint32(coin_state.spent_height),
                                     [],
-                                    uint32(record.wallet_id),
-                                    record.wallet_type,
+                                    pool_wallet.id(),
+                                    WalletType(pool_wallet.type()),
                                     peer,
                                     coin_name,
                                 )
-                            await self.coin_store.set_spent(
-                                curr_coin_state.coin.name(), uint32(curr_coin_state.spent_height)
-                            )
-                            await self.add_interested_coin_ids([new_singleton_coin.name()])
-                            new_coin_state: List[CoinState] = await self.wallet_node.get_coin_state(
-                                [coin_name], peer=peer, fork_height=fork_height
-                            )
-                            assert len(new_coin_state) == 1
-                            curr_coin_state = new_coin_state[0]
-                if record.wallet_type == WalletType.DATA_LAYER:
-                    singleton_spend = await self.wallet_node.fetch_puzzle_solution(
-                        coin_state.spent_height, coin_state.coin, peer
-                    )
-                    dl_wallet = self.wallets[uint32(record.wallet_id)]
-                    await dl_wallet.singleton_removed(
-                        singleton_spend,
-                        coin_state.spent_height,
-                    )
-
-                elif record.wallet_type == WalletType.NFT:
-                    if coin_state.spent_height is not None:
-                        nft_wallet = self.wallets[uint32(record.wallet_id)]
-                        await nft_wallet.remove_coin(coin_state.coin, coin_state.spent_height)
-
-                # Check if a child is a singleton launcher
-                if children is None:
-                    children = await self.wallet_node.fetch_children(coin_name, peer=peer, fork_height=fork_height)
-                assert children is not None
-                for child in children:
-                    if child.coin.puzzle_hash != SINGLETON_LAUNCHER_HASH:
-                        continue
-                    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(
-                        coin_state.spent_height, child.coin, peer
-                    )
-                    if launcher_spend is None:
-                        continue
-                    try:
-                        pool_state = solution_to_pool_state(launcher_spend)
-                        assert pool_state is not None
-                    except (AssertionError, ValueError) as e:
-                        self.log.debug(f"Not a pool wallet launcher {e}")
-                        matched, inner_puzhash = await DataLayerWallet.match_dl_launcher(launcher_spend)
-                        if (
-                            matched
-                            and inner_puzhash is not None
-                            and (await self.puzzle_store.puzzle_hash_exists(inner_puzhash))
-                        ):
-                            for _, wallet in self.wallets.items():
-                                if wallet.type() == WalletType.DATA_LAYER.value:
-                                    dl_wallet = wallet
-                                    break
-                            else:  # No DL wallet exists yet
-                                dl_wallet = await DataLayerWallet.create_new_dl_wallet(
-                                    self,
-                                    self.main_wallet,
-                                )
-                            await dl_wallet.track_new_launcher_id(
-                                child.coin.name(),
-                                peer,
-                                spend=launcher_spend,
-                                height=child.spent_height,
-                            )
-                        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")
-                        continue
-
-                    assert child.spent_height is not None
-                    pool_wallet = await PoolWallet.create(
-                        self,
-                        self.main_wallet,
-                        child.coin.name(),
-                        [launcher_spend],
-                        uint32(child.spent_height),
-                        name="pool_wallet",
-                    )
-                    launcher_spend_additions = launcher_spend.additions()
-                    assert len(launcher_spend_additions) == 1
-                    coin_added = launcher_spend_additions[0]
-                    coin_name = coin_added.name()
-                    existing = await self.coin_store.get_coin_record(coin_name)
-                    if existing is None:
-                        await self.coin_added(
-                            coin_added,
-                            uint32(coin_state.spent_height),
-                            [],
-                            pool_wallet.id(),
-                            WalletType(pool_wallet.type()),
-                            peer,
-                            coin_name,
-                        )
-                    await self.add_interested_coin_ids([coin_name])
+                            await self.add_interested_coin_ids([coin_name])
 
+                    else:
+                        raise RuntimeError("All cases already handled")  # Logic error, all cases handled
+            except Exception as e:
+                tb = traceback.format_exc()
+                self.log.error(f"Error adding state.. {e} {tb}")
+                if retry:
+                    await self.retry_store.add_state(coin_state, peer.peer_node_id, fork_height)
+                continue
             else:
-                raise RuntimeError("All cases already handled")  # Logic error, all cases handled
+                await self.retry_store.remove_state(coin_state)
         for coin_state_removed in trade_coin_removed:
             await self.trade_manager.coins_of_interest_farmed(coin_state_removed, fork_height, peer)
 


Code review comments:
diff_hunk: @@ -932,331 +937,355 @@ async def new_coin_state(
 
         assert len(local_records) == len(coin_states)
         for coin_state, local_record in zip(coin_states, local_records):
-            existing: Optional[WalletCoinRecord]
-            coin_name: bytes32 = coin_state.coin.name()
-            wallet_info: Optional[Tuple[uint32, WalletType]] = await self.get_wallet_id_for_puzzle_hash(
-                coin_state.coin.puzzle_hash
-            )
-            self.log.debug("%s: %s", 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:
-                    local_spent = local_record.spent_block_height
-                if (
-                    local_spent == coin_state.spent_height
-                    and local_record.confirmed_block_height == coin_state.created_height
-                ):
-                    continue
-
-            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.determine_coin_type(peer, coin_state, fork_height)
-                potential_dl = self.get_dl_wallet()
-                if potential_dl is not None:
-                    if await potential_dl.get_singleton_record(coin_state.coin.name()) is not None:
-                        wallet_id = potential_dl.id()
-                        wallet_type = WalletType(potential_dl.type())
-
-            if wallet_id is None or wallet_type is None:
-                self.log.debug(f"No wallet for coin state: {coin_state}")
-                continue
-
-            # Update the DB to signal that we used puzzle hashes up to this one
-            derivation_index = ph_to_index_cache.get(coin_state.coin.puzzle_hash)
-            if derivation_index is None:
-                derivation_index = await self.puzzle_store.index_for_puzzle_hash(coin_state.coin.puzzle_hash)
-            if derivation_index is not None:
-                ph_to_index_cache.put(coin_state.coin.puzzle_hash, derivation_index)
-                if derivation_index > used_up_to:
-                    await self.puzzle_store.set_used_up_to(derivation_index)
-                    used_up_to = max(used_up_to, derivation_index)
-
-            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
-            # if the new coin has not been spent (i.e not ephemeral)
-            elif coin_state.created_height is not None and coin_state.spent_height is None:
-                if local_record is None:
-                    await self.coin_added(
-                        coin_state.coin,
-                        uint32(coin_state.created_height),
-                        all_unconfirmed,
-                        wallet_id,
-                        wallet_type,
-                        peer,
-                        coin_name,
+            try:
+                async with self.db_wrapper.writer():
+                    existing: Optional[WalletCoinRecord]
+                    coin_name: bytes32 = coin_state.coin.name()
+                    wallet_info: Optional[Tuple[uint32, WalletType]] = await self.get_wallet_id_for_puzzle_hash(
+                        coin_state.coin.puzzle_hash
                     )
+                    self.log.debug("%s: %s", coin_name, coin_state)
 
-            # if the coin has been spent
-            elif coin_state.created_height is not None and coin_state.spent_height is not None:
-                self.log.debug("Coin Removed: %s", coin_state)
-                if coin_name in trade_removals:
-                    trade_coin_removed.append(coin_state)
-                children: Optional[List[CoinState]] = None
-                record = local_record
-                if record is None:
-                    farmer_reward = False
-                    pool_reward = False
-                    tx_type: int
-                    if self.is_farmer_reward(uint32(coin_state.created_height), coin_state.coin):
-                        farmer_reward = True
-                        tx_type = TransactionType.FEE_REWARD.value
-                    elif self.is_pool_reward(uint32(coin_state.created_height), coin_state.coin):
-                        pool_reward = True
-                        tx_type = TransactionType.COINBASE_REWARD.value
-                    else:
-                        tx_type = TransactionType.INCOMING_TX.value
-                    record = WalletCoinRecord(
-                        coin_state.coin,
-                        uint32(coin_state.created_height),
-                        uint32(coin_state.spent_height),
-                        True,
-                        farmer_reward or pool_reward,
-                        wallet_type,
-                        wallet_id,
-                    )
-                    await self.coin_store.add_coin_record(record)
-                    # Coin first received
-                    parent_coin_record: Optional[WalletCoinRecord] = await self.coin_store.get_coin_record(
-                        coin_state.coin.parent_coin_info
-                    )
-                    if parent_coin_record is not None and wallet_type.value == parent_coin_record.wallet_type:
-                        change = True
-                    else:
-                        change = False
-
-                    if not change:
-                        created_timestamp = await self.wallet_node.get_timestamp_for_height(coin_state.created_height)
-                        tx_record = TransactionRecord(
-                            confirmed_at_height=uint32(coin_state.created_height),
-                            created_at_time=uint64(created_timestamp),
-                            to_puzzle_hash=(await self.convert_puzzle_hash(wallet_id, coin_state.coin.puzzle_hash)),
-                            amount=uint64(coin_state.coin.amount),
-                            fee_amount=uint64(0),
-                            confirmed=True,
-                            sent=uint32(0),
-                            spend_bundle=None,
-                            additions=[coin_state.coin],
-                            removals=[],
-                            wallet_id=wallet_id,
-                            sent_to=[],
-                            trade_id=None,
-                            type=uint32(tx_type),
-                            name=bytes32(token_bytes()),
-                            memos=[],
-                        )
-                        await self.tx_store.add_transaction_record(tx_record)
-
-                    children = await self.wallet_node.fetch_children(coin_name, peer=peer, fork_height=fork_height)
-                    assert children is not None
-                    additions = [state.coin for state in children]
-                    if len(children) > 0:
-                        fee = 0
-
-                        to_puzzle_hash = None
-                        # Find coin that doesn't belong to us
-                        amount = 0
-                        for coin in additions:
-                            derivation_record = await self.puzzle_store.get_derivation_record_for_puzzle_hash(
-                                coin.puzzle_hash
-                            )
-                            if derivation_record is None:
-                                to_puzzle_hash = coin.puzzle_hash
-                                amount += coin.amount
+                    # If we already have this coin, & it was spent & confirmed at the same heights, then return (done)
+                    if local_record is not None:
+                        local_spent = None
+                        if local_record.spent_block_height != 0:
+                            local_spent = local_record.spent_block_height
+                        if (
+                            local_spent == coin_state.spent_height
+                            and local_record.confirmed_block_height == coin_state.created_height
+                        ):
+                            continue
+
+                    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.determine_coin_type(peer, coin_state, fork_height)
+                        potential_dl = self.get_dl_wallet()
+                        if potential_dl is not None:
+                            if await potential_dl.get_singleton_record(coin_state.coin.name()) is not None:
+                                wallet_id = potential_dl.id()
+                                wallet_type = WalletType(potential_dl.type())
+
+                    if wallet_id is None or wallet_type is None:
+                        self.log.debug(f"No wallet for coin state: {coin_state}")
+                        continue
 
-                        if to_puzzle_hash is None:
-                            to_puzzle_hash = additions[0].puzzle_hash
+                    # Update the DB to signal that we used puzzle hashes up to this one
+                    derivation_index = ph_to_index_cache.get(coin_state.coin.puzzle_hash)
+                    if derivation_index is None:
+                        derivation_index = await self.puzzle_store.index_for_puzzle_hash(coin_state.coin.puzzle_hash)
+                    if derivation_index is not None:
+                        ph_to_index_cache.put(coin_state.coin.puzzle_hash, derivation_index)
+                        if derivation_index > used_up_to:
+                            await self.puzzle_store.set_used_up_to(derivation_index)
+                            used_up_to = max(used_up_to, derivation_index)
+
+                    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
+                    # if the new coin has not been spent (i.e not ephemeral)
+                    elif coin_state.created_height is not None and coin_state.spent_height is None:
+                        if local_record is None:
+                            await self.coin_added(
+                                coin_state.coin,
+                                uint32(coin_state.created_height),
+                                all_unconfirmed,
+                                wallet_id,
+                                wallet_type,
+                                peer,
+                                coin_name,
+                            )
 
-                        spent_timestamp = await self.wallet_node.get_timestamp_for_height(coin_state.spent_height)
+                    # if the coin has been spent
+                    elif coin_state.created_height is not None and coin_state.spent_height is not None:
+                        self.log.debug("Coin Removed: %s", coin_state)
+                        if coin_name in trade_removals:
+                            trade_coin_removed.append(coin_state)
+                        children: Optional[List[CoinState]] = None
+                        record = local_record
+                        if record is None:
+                            farmer_reward = False
+                            pool_reward = False
+                            tx_type: int
+                            if self.is_farmer_reward(uint32(coin_state.created_height), coin_state.coin):
+                                farmer_reward = True
+                                tx_type = TransactionType.FEE_REWARD.value
+                            elif self.is_pool_reward(uint32(coin_state.created_height), coin_state.coin):
+                                pool_reward = True
+                                tx_type = TransactionType.COINBASE_REWARD.value
+                            else:
+                                tx_type = TransactionType.INCOMING_TX.value
+                            record = WalletCoinRecord(
+                                coin_state.coin,
+                                uint32(coin_state.created_height),
+                                uint32(coin_state.spent_height),
+                                True,
+                                farmer_reward or pool_reward,
+                                wallet_type,
+                                wallet_id,
+                            )
+                            await self.coin_store.add_coin_record(record)
+                            # Coin first received
+                            parent_coin_record: Optional[WalletCoinRecord] = await self.coin_store.get_coin_record(
+                                coin_state.coin.parent_coin_info
+                            )
+                            if parent_coin_record is not None and wallet_type.value == parent_coin_record.wallet_type:
+                                change = True
+                            else:
+                                change = False
+
+                            if not change:
+                                created_timestamp = await self.wallet_node.get_timestamp_for_height(
+                                    coin_state.created_height
+                                )
+                                tx_record = TransactionRecord(
+                                    confirmed_at_height=uint32(coin_state.created_height),
+                                    created_at_time=uint64(created_timestamp),
+                                    to_puzzle_hash=(
+                                        await self.convert_puzzle_hash(wallet_id, coin_state.coin.puzzle_hash)
+                                    ),
+                                    amount=uint64(coin_state.coin.amount),
+                                    fee_amount=uint64(0),
+                                    confirmed=True,
+                                    sent=uint32(0),
+                                    spend_bundle=None,
+                                    additions=[coin_state.coin],
+                                    removals=[],
+                                    wallet_id=wallet_id,
+                                    sent_to=[],
+                                    trade_id=None,
+                                    type=uint32(tx_type),
+                                    name=bytes32(token_bytes()),
+                                    memos=[],
+                                )
+                                await self.tx_store.add_transaction_record(tx_record)
 
-                        # Reorg rollback adds reorged transactions so it's possible there is tx_record already
-                        # Even though we are just adding coin record to the db (after reorg)
-                        tx_records: List[TransactionRecord] = []
-                        for out_tx_record in all_unconfirmed:
-                            for rem_coin in out_tx_record.removals:
-                                if rem_coin == coin_state.coin:
-                                    tx_records.append(out_tx_record)
+                            children = await self.wallet_node.fetch_children(
+                                coin_name, peer=peer, fork_height=fork_height
+                            )
+                            assert children is not None
+                            additions = [state.coin for state in children]
+                            if len(children) > 0:
+                                fee = 0
+
+                                to_puzzle_hash = None
+                                # Find coin that doesn't belong to us
+                                amount = 0
+                                for coin in additions:
+                                    derivation_record = await self.puzzle_store.get_derivation_record_for_puzzle_hash(
+                                        coin.puzzle_hash
+                                    )
+                                    if derivation_record is None:
+                                        to_puzzle_hash = coin.puzzle_hash
+                                        amount += coin.amount
+
+                                if to_puzzle_hash is None:
+                                    to_puzzle_hash = additions[0].puzzle_hash
+
+                                spent_timestamp = await self.wallet_node.get_timestamp_for_height(
+                                    coin_state.spent_height
+                                )
 
-                        if len(tx_records) > 0:
-                            for tx_record in tx_records:
-                                await self.tx_store.set_confirmed(tx_record.name, uint32(coin_state.spent_height))
+                                # Reorg rollback adds reorged transactions so it's possible there is tx_record already
+                                # Even though we are just adding coin record to the db (after reorg)
+                                tx_records: List[TransactionRecord] = []
+                                for out_tx_record in all_unconfirmed:
+                                    for rem_coin in out_tx_record.removals:
+                                        if rem_coin == coin_state.coin:
+                                            tx_records.append(out_tx_record)
+
+                                if len(tx_records) > 0:
+                                    for tx_record in tx_records:
+                                        await self.tx_store.set_confirmed(
+                                            tx_record.name, uint32(coin_state.spent_height)
+                                        )
+                                else:
+                                    tx_record = TransactionRecord(
+                                        confirmed_at_height=uint32(coin_state.spent_height),
+                                        created_at_time=uint64(spent_timestamp),
+                                        to_puzzle_hash=(await self.convert_puzzle_hash(wallet_id, to_puzzle_hash)),
+                                        amount=uint64(int(amount)),
+                                        fee_amount=uint64(fee),
+                                        confirmed=True,
+                                        sent=uint32(0),
+                                        spend_bundle=None,
+                                        additions=additions,
+                                        removals=[coin_state.coin],
+                                        wallet_id=wallet_id,
+                                        sent_to=[],
+                                        trade_id=None,
+                                        type=uint32(TransactionType.OUTGOING_TX.value),
+                                        name=bytes32(token_bytes()),
+                                        memos=[],
+                                    )
+
+                                    await self.tx_store.add_transaction_record(tx_record)
                         else:
-                            tx_record = TransactionRecord(
-                                confirmed_at_height=uint32(coin_state.spent_height),
-                                created_at_time=uint64(spent_timestamp),
-                                to_puzzle_hash=(await self.convert_puzzle_hash(wallet_id, to_puzzle_hash)),
-                                amount=uint64(int(amount)),
-                                fee_amount=uint64(fee),
-                                confirmed=True,
-                                sent=uint32(0),
-                                spend_bundle=None,
-                                additions=additions,
-                                removals=[coin_state.coin],
-                                wallet_id=wallet_id,
-                                sent_to=[],
-                                trade_id=None,
-                                type=uint32(TransactionType.OUTGOING_TX.value),
-                                name=bytes32(token_bytes()),
-                                memos=[],
+                            await self.coin_store.set_spent(coin_name, uint32(coin_state.spent_height))
+                            rem_tx_records: List[TransactionRecord] = []
+                            for out_tx_record in all_unconfirmed:
+                                for rem_coin in out_tx_record.removals:
+                                    if rem_coin == coin_state.coin:
+                                        rem_tx_records.append(out_tx_record)
+
+                            for tx_record in rem_tx_records:
+                                await self.tx_store.set_confirmed(tx_record.name, uint32(coin_state.spent_height))
+                        for unconfirmed_record in all_unconfirmed:
+                            for rem_coin in unconfirmed_record.removals:
+                                if rem_coin == coin_state.coin:
+                                    self.log.info(f"Setting tx_id: {unconfirmed_record.name} to confirmed")
+                                    await self.tx_store.set_confirmed(
+                                        unconfirmed_record.name, uint32(coin_state.spent_height)
+                                    )
+
+                        if record.wallet_type == WalletType.POOLING_WALLET:
+                            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(
+                                        curr_coin_state.spent_height, curr_coin_state.coin, peer
+                                    )
+                                    success = await wallet.apply_state_transition(cs, curr_coin_state.spent_height)
+                                    if not success:
+                                        break
+                                    new_singleton_coin: Optional[Coin] = wallet.get_next_interesting_coin(cs)
+                                    if new_singleton_coin is None:
+                                        # No more singleton (maybe destroyed?)
+                                        break
+
+                                    coin_name = new_singleton_coin.name()
+                                    existing = await self.coin_store.get_coin_record(coin_name)
+                                    if existing is None:
+                                        await self.coin_added(
+                                            new_singleton_coin,
+                                            uint32(coin_state.spent_height),
+                                            [],
+                                            uint32(record.wallet_id),
+                                            record.wallet_type,
+                                            peer,
+                                            coin_name,
+                                        )
+                                    await self.coin_store.set_spent(
+                                        curr_coin_state.coin.name(), uint32(curr_coin_state.spent_height)
+                                    )
+                                    await self.add_interested_coin_ids([new_singleton_coin.name()])
+                                    new_coin_state: List[CoinState] = await self.wallet_node.get_coin_state(
+                                        [coin_name], peer=peer, fork_height=fork_height
+                                    )
+                                    assert len(new_coin_state) == 1
+                                    curr_coin_state = new_coin_state[0]
+                        if record.wallet_type == WalletType.DATA_LAYER:
+                            singleton_spend = await self.wallet_node.fetch_puzzle_solution(
+                                coin_state.spent_height, coin_state.coin, peer
+                            )
+                            dl_wallet = self.wallets[uint32(record.wallet_id)]
+                            await dl_wallet.singleton_removed(
+                                singleton_spend,
+                                coin_state.spent_height,
                             )
 
-                            await self.tx_store.add_transaction_record(tx_record)
-                else:
-                    await self.coin_store.set_spent(coin_name, uint32(coin_state.spent_height))
-                    rem_tx_records: List[TransactionRecord] = []
-                    for out_tx_record in all_unconfirmed:
-                        for rem_coin in out_tx_record.removals:
-                            if rem_coin == coin_state.coin:
-                                rem_tx_records.append(out_tx_record)
-
-                    for tx_record in rem_tx_records:
-                        await self.tx_store.set_confirmed(tx_record.name, uint32(coin_state.spent_height))
-                for unconfirmed_record in all_unconfirmed:
-                    for rem_coin in unconfirmed_record.removals:
-                        if rem_coin == coin_state.coin:
-                            self.log.info(f"Setting tx_id: {unconfirmed_record.name} to confirmed")
-                            await self.tx_store.set_confirmed(unconfirmed_record.name, uint32(coin_state.spent_height))
-
-                if record.wallet_type == WalletType.POOLING_WALLET:
-                    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(
-                                curr_coin_state.spent_height, curr_coin_state.coin, peer
+                        elif record.wallet_type == WalletType.NFT:
+                            if coin_state.spent_height is not None:
+                                nft_wallet = self.wallets[uint32(record.wallet_id)]
+                                await nft_wallet.remove_coin(coin_state.coin, coin_state.spent_height)
+
+                        # Check if a child is a singleton launcher
+                        if children is None:
+                            children = await self.wallet_node.fetch_children(
+                                coin_name, peer=peer, fork_height=fork_height
+                            )
+                        assert children is not None
+                        for child in children:
+                            if child.coin.puzzle_hash != SINGLETON_LAUNCHER_HASH:
+                                continue
+                            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(
+                                coin_state.spent_height, child.coin, peer
+                            )
+                            if launcher_spend is None:
+                                continue
+                            try:
+                                pool_state = solution_to_pool_state(launcher_spend)
+                                assert pool_state is not None
+                            except (AssertionError, ValueError) as e:
+                                self.log.debug(f"Not a pool wallet launcher {e}")
+                                matched, inner_puzhash = await DataLayerWallet.match_dl_launcher(launcher_spend)
+                                if (
+                                    matched
+                                    and inner_puzhash is not None
+                                    and (await self.puzzle_store.puzzle_hash_exists(inner_puzhash))
+                                ):
+                                    for _, wallet in self.wallets.items():
+                                        if wallet.type() == WalletType.DATA_LAYER.value:
+                                            dl_wallet = wallet
+                                            break
+                                    else:  # No DL wallet exists yet
+                                        dl_wallet = await DataLayerWallet.create_new_dl_wallet(
+                                            self,
+                                            self.main_wallet,
+                                        )
+                                    await dl_wallet.track_new_launcher_id(
+                                        child.coin.name(),
+                                        peer,
+                                        spend=launcher_spend,
+                                        height=child.spent_height,
+                                    )
+                                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")
+                                continue
+
+                            assert child.spent_height is not None
+                            pool_wallet = await PoolWallet.create(
+                                self,
+                                self.main_wallet,
+                                child.coin.name(),
+                                [launcher_spend],
+                                uint32(child.spent_height),
+                                name="pool_wallet",
                             )
-                            success = await wallet.apply_state_transition(cs, curr_coin_state.spent_height)
-                            if not success:
-                                break
-                            new_singleton_coin: Optional[Coin] = wallet.get_next_interesting_coin(cs)
-                            if new_singleton_coin is None:
-                                # No more singleton (maybe destroyed?)
-                                break
-
-                            coin_name = new_singleton_coin.name()
+                            launcher_spend_additions = launcher_spend.additions()
+                            assert len(launcher_spend_additions) == 1
+                            coin_added = launcher_spend_additions[0]
+                            coin_name = coin_added.name()
                             existing = await self.coin_store.get_coin_record(coin_name)
                             if existing is None:
                                 await self.coin_added(
-                                    new_singleton_coin,
+                                    coin_added,
                                     uint32(coin_state.spent_height),
                                     [],
-                                    uint32(record.wallet_id),
-                                    record.wallet_type,
+                                    pool_wallet.id(),
+                                    WalletType(pool_wallet.type()),
                                     peer,
                                     coin_name,
                                 )
-                            await self.coin_store.set_spent(
-                                curr_coin_state.coin.name(), uint32(curr_coin_state.spent_height)
-                            )
-                            await self.add_interested_coin_ids([new_singleton_coin.name()])
-                            new_coin_state: List[CoinState] = await self.wallet_node.get_coin_state(
-                                [coin_name], peer=peer, fork_height=fork_height
-                            )
-                            assert len(new_coin_state) == 1
-                            curr_coin_state = new_coin_state[0]
-                if record.wallet_type == WalletType.DATA_LAYER:
-                    singleton_spend = await self.wallet_node.fetch_puzzle_solution(
-                        coin_state.spent_height, coin_state.coin, peer
-                    )
-                    dl_wallet = self.wallets[uint32(record.wallet_id)]
-                    await dl_wallet.singleton_removed(
-                        singleton_spend,
-                        coin_state.spent_height,
-                    )
-
-                elif record.wallet_type == WalletType.NFT:
-                    if coin_state.spent_height is not None:
-                        nft_wallet = self.wallets[uint32(record.wallet_id)]
-                        await nft_wallet.remove_coin(coin_state.coin, coin_state.spent_height)
-
-                # Check if a child is a singleton launcher
-                if children is None:
-                    children = await self.wallet_node.fetch_children(coin_name, peer=peer, fork_height=fork_height)
-                assert children is not None
-                for child in children:
-                    if child.coin.puzzle_hash != SINGLETON_LAUNCHER_HASH:
-                        continue
-                    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(
-                        coin_state.spent_height, child.coin, peer
-                    )
-                    if launcher_spend is None:
-                        continue
-                    try:
-                        pool_state = solution_to_pool_state(launcher_spend)
-                        assert pool_state is not None
-                    except (AssertionError, ValueError) as e:
-                        self.log.debug(f"Not a pool wallet launcher {e}")
-                        matched, inner_puzhash = await DataLayerWallet.match_dl_launcher(launcher_spend)
-                        if (
-                            matched
-                            and inner_puzhash is not None
-                            and (await self.puzzle_store.puzzle_hash_exists(inner_puzhash))
-                        ):
-                            for _, wallet in self.wallets.items():
-                                if wallet.type() == WalletType.DATA_LAYER.value:
-                                    dl_wallet = wallet
-                                    break
-                            else:  # No DL wallet exists yet
-                                dl_wallet = await DataLayerWallet.create_new_dl_wallet(
-                                    self,
-                                    self.main_wallet,
-                                )
-                            await dl_wallet.track_new_launcher_id(
-                                child.coin.name(),
-                                peer,
-                                spend=launcher_spend,
-                                height=child.spent_height,
-                            )
-                        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")
-                        continue
-
-                    assert child.spent_height is not None
-                    pool_wallet = await PoolWallet.create(
-                        self,
-                        self.main_wallet,
-                        child.coin.name(),
-                        [launcher_spend],
-                        uint32(child.spent_height),
-                        name="pool_wallet",
-                    )
-                    launcher_spend_additions = launcher_spend.additions()
-                    assert len(launcher_spend_additions) == 1
-                    coin_added = launcher_spend_additions[0]
-                    coin_name = coin_added.name()
-                    existing = await self.coin_store.get_coin_record(coin_name)
-                    if existing is None:
-                        await self.coin_added(
-                            coin_added,
-                            uint32(coin_state.spent_height),
-                            [],
-                            pool_wallet.id(),
-                            WalletType(pool_wallet.type()),
-                            peer,
-                            coin_name,
-                        )
-                    await self.add_interested_coin_ids([coin_name])
+                            await self.add_interested_coin_ids([coin_name])
 
+                    else:
+                        raise RuntimeError("All cases already handled")  # Logic error, all cases handled
+            except Exception as e:
+                tb = traceback.format_exc()
+                self.log.error(f"Error adding state.. {e} {tb}")
path: chia/wallet/wallet_state_manager.py
text: @user1:
lets change this to `self.log.exception`

diff_hunk: @@ -377,6 +382,42 @@ async def _messages_to_resend(self) -> List[Tuple[Message, Set[bytes32]]]:
 
         return messages
 
+    async def _retry_failed_states(self):
+        while not self._shut_down:
+            try:
+                await asyncio.sleep(5)
+                current_time = time.time()
+                if self.last_state_retry_time < current_time - 10:
+                    self.last_state_retry_time = current_time
+                    if self.wallet_state_manager is None:
+                        continue
+                    states_to_retry = await self.wallet_state_manager.retry_store.get_all_states_to_retry()
+                    for state, peer_id, fork_height in states_to_retry:
+                        matching_peer = tuple(
+                            p for p in self.server.get_full_node_connections() if p.peer_node_id == peer_id
+                        )
+                        if len(matching_peer) == 0:
+                            peer = self.get_full_node_peer()
+                            self.log.info(
+                                f"disconnected from peer {peer_id}, state will retry with {peer.peer_node_id}"
+                            )
+                        else:
+                            peer = matching_peer[0]
+                        async with self.wallet_state_manager.db_wrapper.writer():
+                            self.log.info(f"retrying coin_state: {state}")
+                            try:
+                                await self.wallet_state_manager.new_coin_state(
+                                    [state], peer, None if fork_height == 0 else fork_height, False
+                                )
+                            except Exception as e:
+                                tb = traceback.format_exc()
+                                self.log.error(f"Exception while adding states.. : {e} {tb}")
path: chia/wallet/wallet_node.py
text: @user1:
`self.log.exception`