better handling of offer status and failed txs (#14812)

* better handling of offer status and failed txs

* 5 retries before tx is marked failed

* increase test timeouts

* fixed test

* require only difference vs whole fee

* polish + fee coins kick in only if we're missing fee

chia/wallet/trade_manager.py

@@ -210,6 +210,10 @@ class TradeManager:
         await self.trade_store.set_status(trade_id, TradeStatus.CANCELLED)
         self.wallet_state_manager.state_changed("offer_cancelled")
 
+    async def fail_pending_offer(self, trade_id: bytes32) -> None:
+        await self.trade_store.set_status(trade_id, TradeStatus.FAILED)
+        self.wallet_state_manager.state_changed("offer_failed")
+
     async def cancel_pending_offer_safely(
         self, trade_id: bytes32, fee: uint64 = uint64(0)
     ) -> Optional[List[TransactionRecord]]:
@@ -761,7 +765,7 @@ class TradeManager:
         success, take_offer, error = result
 
         complete_offer = await self.check_for_final_modifications(Offer.aggregate([offer, take_offer]), solver)
-        self.log.info(f"COMPLETE OFFER: {complete_offer.to_bech32()}")
+        self.log.info("COMPLETE OFFER: %s", complete_offer.to_bech32())
         assert complete_offer.is_valid()
         final_spend_bundle: SpendBundle = complete_offer.to_valid_spend()
         await self.maybe_create_wallets_for_offer(complete_offer)

chia/wallet/transaction_record.py

@@ -9,6 +9,7 @@ from chia.types.blockchain_format.sized_bytes import bytes32
 from chia.types.mempool_inclusion_status import MempoolInclusionStatus
 from chia.types.spend_bundle import SpendBundle
 from chia.util.bech32m import decode_puzzle_hash, encode_puzzle_hash
+from chia.util.errors import Err
 from chia.util.ints import uint8, uint32, uint64
 from chia.util.streamable import Streamable, streamable
 from chia.wallet.util.transaction_type import TransactionType
@@ -109,3 +110,16 @@ class TransactionRecord(Streamable):
             if memo is not None
         }
         return formatted
+
+    def is_valid(self) -> bool:
+        past_receipts = self.sent_to
+        if len(past_receipts) < 6:
+            # we haven't tried enough peers yet
+            return True
+        if any([x[0] for x in past_receipts if x[0] == MempoolInclusionStatus.SUCCESS.value]):
+            # we managed to push it to mempool at least once
+            return True
+        if any([x[1] for x in past_receipts if x[1] in (Err.INVALID_FEE_LOW_FEE, Err.INVALID_FEE_TOO_CLOSE_TO_ZERO)]):
+            # we tried to push it to mempool and got a fee error so it's a temporary error
+            return True
+        return False

chia/wallet/wallet.py

@@ -50,7 +50,6 @@ from chia.wallet.wallet_info import WalletInfo
 if TYPE_CHECKING:
     from chia.server.ws_connection import WSChiaConnection
 
-
 # https://github.com/Chia-Network/chips/blob/80e4611fe52b174bf1a0382b9dff73805b18b8c6/CHIPs/chip-0002.md#signmessage
 CHIP_0002_SIGN_MESSAGE_PREFIX = "Chia Signed Message"
 
@@ -352,7 +351,7 @@ class Wallet:
         if not ignore_max_send_amount:
             max_send = await self.get_max_send_amount()
             if total_amount > max_send:
-                raise ValueError(f"Can't send more than {max_send} mojos in a single transaction")
+                raise ValueError(f"Can't send more than {max_send} mojos in a single transaction, got {total_amount}")
             self.log.debug("Got back max send amount: %s", max_send)
         if coins is None:
             if total_amount > total_balance:
@@ -371,15 +370,27 @@ class Wallet:
             )
         elif exclude_coins is not None:
             raise ValueError("Can't exclude coins when also specifically including coins")
+
         assert len(coins) > 0
         self.log.info(f"coins is not None {coins}")
         spend_value = sum([coin.amount for coin in coins])
-
+        self.log.info(f"spend_value is {spend_value} and total_amount is {total_amount}")
         change = spend_value - total_amount
         if negative_change_allowed:
             change = max(0, change)
-
-        assert change >= 0
+        # only kicks in if fee is missing
+        if change < 0 and fee + amount == total_amount:
+            fee_coins = await self.select_coins(
+                # change already includes fee amount
+                uint64(abs(change)),
+                excluded_coin_amounts=exclude_coin_amounts,
+                exclude=([] if exclude_coins is None else list(exclude_coins)) + list(coins or []),
+            )
+            coins = coins.union(fee_coins)
+            spend_value = sum([coin.amount for coin in coins])
+            self.log.info(f"Updated spend_value is {spend_value} and total_amount is {total_amount}")
+            change = spend_value - total_amount
+        assert change >= 0, f"change is negative: {change}"
 
         if coin_announcements_to_consume is not None:
             coin_announcements_bytes: Optional[Set[bytes32]] = {a.name() for a in coin_announcements_to_consume}

chia/wallet/wallet_state_manager.py

@@ -1030,7 +1030,7 @@ class WalletStateManager:
 
         trade_removals = await self.trade_manager.get_coins_of_interest()
         all_unconfirmed: List[TransactionRecord] = await self.tx_store.get_all_unconfirmed()
-        trade_coin_removed: List[CoinState] = []
+        trade_coin_removed: Set[CoinState] = set([])
         used_up_to = -1
         ph_to_index_cache: LRUCache = LRUCache(100)
 
@@ -1066,7 +1066,8 @@ class WalletStateManager:
                             continue
 
                     if coin_state.spent_height is not None and coin_name in trade_removals:
-                        trade_coin_removed.append(coin_state)
+                        trade_coin_removed.add(coin_state)
+
                     wallet_id: Optional[uint32] = None
                     wallet_type: Optional[WalletType] = None
                     if wallet_info is not None:
@@ -1597,28 +1598,43 @@ class WalletStateManager:
         updated = await self.tx_store.increment_sent(spendbundle_id, name, send_status, error)
         if updated:
             tx: Optional[TransactionRecord] = await self.get_transaction(spendbundle_id)
-            if tx is not None:
+            if tx is not None and tx.spend_bundle is not None:
+                self.log.info("Checking if we need to cancel trade for tx: %s", tx.name)
                 # we're only interested in errors that are not temporary
-                if send_status and error and error not in (Err.INVALID_FEE_LOW_FEE, Err.INVALID_FEE_TOO_CLOSE_TO_ZERO):
-                    if tx.spend_bundle is not None:
-                        coins_removed = tx.spend_bundle.removals()
-                        trade_coins_removed = set([])
-                        for removed_coin in coins_removed:
-                            trade = await self.trade_manager.get_trade_by_coin(removed_coin)
-                            if trade is not None and trade.status in (
-                                TradeStatus.PENDING_CONFIRM.value,
-                                TradeStatus.PENDING_ACCEPT.value,
-                                TradeStatus.PENDING_CANCEL.value,
-                            ):
-                                # offer was tied to these coins, lets subscribe to them to get a confirmation to
-                                # cancel it if it's confirmed
-                                # we send transactions to multiple peers, and in cases when mempool gets
-                                # fragmented, it's safest to wait for confirmation from blockchain before setting
-                                # offer to failed
-                                trade_coins_removed.add(removed_coin.name())
-                        if trade_coins_removed:
+                if (
+                    send_status != MempoolInclusionStatus.SUCCESS
+                    and error
+                    and error not in (Err.INVALID_FEE_LOW_FEE, Err.INVALID_FEE_TOO_CLOSE_TO_ZERO)
+                ):
+                    coins_removed = tx.spend_bundle.removals()
+                    trade_coins_removed = set([])
+                    trade = None
+                    for removed_coin in coins_removed:
+                        trade = await self.trade_manager.get_trade_by_coin(removed_coin)
+                        if trade is not None and trade.status in (
+                            TradeStatus.PENDING_CONFIRM.value,
+                            TradeStatus.PENDING_ACCEPT.value,
+                            TradeStatus.PENDING_CANCEL.value,
+                        ):
+                            # offer was tied to these coins, lets subscribe to them to get a confirmation to
+                            # cancel it if it's confirmed
+                            # we send transactions to multiple peers, and in cases when mempool gets
+                            # fragmented, it's safest to wait for confirmation from blockchain before setting
+                            # offer to failed
+                            trade_coins_removed.add(removed_coin.name())
+                    if trade and trade_coins_removed:
+                        if not tx.is_valid():
+                            # we've tried to send this transaction to a full node multiple times
+                            # but failed, it's safe to assume that it's not going to be accepted
+                            # we can mark this offer as failed
+                            self.log.info("This offer can't be posted, removing it from pending offers")
+                            assert trade is not None
+                            await self.trade_manager.fail_pending_offer(trade.trade_id)
+
+                        else:
                             self.log.info(
-                                "Subscribing to unspendable offer coins: %s", [x.hex() for x in trade_coins_removed]
+                                "Subscribing to unspendable offer coins: %s",
+                                [x.hex() for x in trade_coins_removed],
                             )
                             await self.add_interested_coin_ids(list(trade_coins_removed))
 

chia/wallet/wallet_transaction_store.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import dataclasses
+import logging
 import time
 from typing import Dict, List, Optional, Tuple
 
@@ -13,6 +14,8 @@ from chia.wallet.transaction_record import TransactionRecord
 from chia.wallet.transaction_sorting import SortKey
 from chia.wallet.util.transaction_type import TransactionType
 
+log = logging.getLogger(__name__)
+
 
 def filter_ok_mempool_status(sent_to: List[Tuple[str, uint8, Optional[str]]]) -> List[Tuple[str, uint8, Optional[str]]]:
     """Remove SUCCESS and PENDING status records from a TransactionRecord sent_to field"""
@@ -152,6 +155,10 @@ class WalletTransactionStore:
         sent_to.append(append_data)
 
         tx: TransactionRecord = dataclasses.replace(current, sent=sent_count, sent_to=sent_to)
+        if not tx.is_valid():
+            # if the tx is not valid due to repeated failures, we will confirm that we can't spend it
+            log.info(f"Marking tx={tx.name} as confirmed but failed, since it is not spendable due to errors")
+            tx = dataclasses.replace(tx, confirmed=True, confirmed_at_height=uint32(0))
         await self.add_transaction_record(tx)
         return True
 

tests/wallet/cat_wallet/test_trades.py

@@ -1,11 +1,13 @@
 from __future__ import annotations
 
 import asyncio
+import dataclasses
 import sys
 from secrets import token_bytes
 from typing import Dict, List
 
 import pytest
+from blspy import G2Element
 
 from chia.consensus.cost_calculator import NPCResult
 from chia.full_node.bundle_tools import simple_solution_generator
@@ -791,11 +793,117 @@ class TestCATTrades:
         with pytest.raises(ValueError):
             await trade_manager_taker.respond_to_offer(offer, peer, fee=uint64(10))
         await time_out_assert(15, get_trade_and_status, TradeStatus.PENDING_CONFIRM, trade_manager_taker, tr1)
-
         # pushing into mempool while already in it should fail
         tr2, txs2 = await trade_manager_trader.respond_to_offer(offer, peer, fee=uint64(10))
-        await time_out_assert(15, get_trade_and_status, TradeStatus.PENDING_CONFIRM, trade_manager_trader, tr2)
         assert await trade_manager_trader.get_coins_of_interest()
         offer_tx_records: List[TransactionRecord] = await wallet_node_maker.wallet_state_manager.tx_store.get_not_sent()
         await full_node.process_transaction_records(records=offer_tx_records)
         await time_out_assert(15, get_trade_and_status, TradeStatus.FAILED, trade_manager_trader, tr2)
+
+    @pytest.mark.asyncio
+    async def test_trade_bad_spend(self, wallets_prefarm):
+        (
+            [wallet_node_maker, maker_funds],
+            [wallet_node_taker, taker_funds],
+            full_node,
+        ) = wallets_prefarm
+        wallet_maker = wallet_node_maker.wallet_state_manager.main_wallet
+        xch_to_cat_amount = uint64(100)
+
+        async with wallet_node_maker.wallet_state_manager.lock:
+            cat_wallet_maker: CATWallet = await CATWallet.create_new_cat_wallet(
+                wallet_node_maker.wallet_state_manager, wallet_maker, {"identifier": "genesis_by_id"}, xch_to_cat_amount
+            )
+
+            tx_records: List[TransactionRecord] = await wallet_node_maker.wallet_state_manager.tx_store.get_not_sent()
+
+        await full_node.process_transaction_records(records=tx_records)
+
+        await time_out_assert(15, cat_wallet_maker.get_confirmed_balance, xch_to_cat_amount)
+        await time_out_assert(15, cat_wallet_maker.get_unconfirmed_balance, xch_to_cat_amount)
+        maker_funds -= xch_to_cat_amount
+        await time_out_assert(15, wallet_maker.get_confirmed_balance, maker_funds)
+
+        chia_for_cat = {
+            wallet_maker.id(): 1000,
+            cat_wallet_maker.id(): -4,
+        }
+
+        trade_manager_maker = wallet_node_maker.wallet_state_manager.trade_manager
+        trade_manager_taker = wallet_node_taker.wallet_state_manager.trade_manager
+
+        async def get_trade_and_status(trade_manager, trade) -> TradeStatus:
+            trade_rec = await trade_manager.get_trade_by_id(trade.trade_id)
+            if trade_rec:
+                return TradeStatus(trade_rec.status)
+            raise ValueError("Couldn't find the trade record")
+
+        success, trade_make, error = await trade_manager_maker.create_offer_for_ids(chia_for_cat)
+        await time_out_assert(10, get_trade_and_status, TradeStatus.PENDING_ACCEPT, trade_manager_maker, trade_make)
+        assert error is None
+        assert success is True
+        assert trade_make is not None
+        peer = wallet_node_taker.get_full_node_peer()
+        offer = Offer.from_bytes(trade_make.offer)
+        bundle = dataclasses.replace(offer._bundle, aggregated_signature=G2Element())
+        offer = dataclasses.replace(offer, _bundle=bundle)
+        tr1, txs1 = await trade_manager_taker.respond_to_offer(offer, peer, fee=uint64(10))
+        wallet_node_taker.wallet_tx_resend_timeout_secs = 1  # don't wait for resend
+        await wallet_node_taker._resend_queue()
+        await wallet_node_taker._resend_queue()
+        await wallet_node_taker._resend_queue()
+        await wallet_node_taker._resend_queue()
+        await wallet_node_taker._resend_queue()
+        await wallet_node_taker._resend_queue()
+        offer_tx_records: List[TransactionRecord] = await wallet_node_maker.wallet_state_manager.tx_store.get_not_sent()
+        await full_node.process_transaction_records(records=offer_tx_records)
+        await time_out_assert(15, get_trade_and_status, TradeStatus.FAILED, trade_manager_taker, tr1)
+
+    @pytest.mark.asyncio
+    async def test_trade_high_fee(self, wallets_prefarm):
+        (
+            [wallet_node_maker, maker_funds],
+            [wallet_node_taker, taker_funds],
+            full_node,
+        ) = wallets_prefarm
+        wallet_maker = wallet_node_maker.wallet_state_manager.main_wallet
+        xch_to_cat_amount = uint64(100)
+
+        async with wallet_node_maker.wallet_state_manager.lock:
+            cat_wallet_maker: CATWallet = await CATWallet.create_new_cat_wallet(
+                wallet_node_maker.wallet_state_manager, wallet_maker, {"identifier": "genesis_by_id"}, xch_to_cat_amount
+            )
+
+            tx_records: List[TransactionRecord] = await wallet_node_maker.wallet_state_manager.tx_store.get_not_sent()
+
+        await full_node.process_transaction_records(records=tx_records)
+
+        await time_out_assert(15, cat_wallet_maker.get_confirmed_balance, xch_to_cat_amount)
+        await time_out_assert(15, cat_wallet_maker.get_unconfirmed_balance, xch_to_cat_amount)
+        maker_funds -= xch_to_cat_amount
+        await time_out_assert(15, wallet_maker.get_confirmed_balance, maker_funds)
+
+        chia_for_cat = {
+            wallet_maker.id(): 1000,
+            cat_wallet_maker.id(): -4,
+        }
+
+        trade_manager_maker = wallet_node_maker.wallet_state_manager.trade_manager
+        trade_manager_taker = wallet_node_taker.wallet_state_manager.trade_manager
+
+        async def get_trade_and_status(trade_manager, trade) -> TradeStatus:
+            trade_rec = await trade_manager.get_trade_by_id(trade.trade_id)
+            if trade_rec:
+                return TradeStatus(trade_rec.status)
+            raise ValueError("Couldn't find the trade record")
+
+        success, trade_make, error = await trade_manager_maker.create_offer_for_ids(chia_for_cat)
+        await time_out_assert(10, get_trade_and_status, TradeStatus.PENDING_ACCEPT, trade_manager_maker, trade_make)
+        assert error is None
+        assert success is True
+        assert trade_make is not None
+        peer = wallet_node_taker.get_full_node_peer()
+        offer = Offer.from_bytes(trade_make.offer)
+        tr1, txs1 = await trade_manager_taker.respond_to_offer(offer, peer, fee=uint64(1000000000000))
+        await full_node.process_transaction_records(records=txs1)
+        await time_out_assert(15, get_trade_and_status, TradeStatus.CONFIRMED, trade_manager_taker, tr1)