[LABS-293] Ensure wallet is synced for RPC TX endpoints (#20190)

* Ensure wallet is synced for RPC tx endpoints

* Address flakes

* more flakes

* More flakes in DL tests

Author: Quexington

chia/_tests/core/data_layer/test_data_rpc.py

@@ -73,7 +73,6 @@
 from chia.wallet.trading.offer import Offer as TradingOffer
 from chia.wallet.transaction_record import TransactionRecord
 from chia.wallet.util.tx_config import DEFAULT_TX_CONFIG
-from chia.wallet.wallet import Wallet
 from chia.wallet.wallet_node import WalletNode
 from chia.wallet.wallet_request_types import CheckOfferValidity, DLLatestSingleton
 from chia.wallet.wallet_rpc_api import WalletRpcApi
@@ -763,6 +762,7 @@ async def test_get_owned_stores(
         ph = await action_scope.get_puzzle_hash(wallet_node.wallet_state_manager)
     for i in range(num_blocks):
         await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(ph))
+    await full_node_api.wait_for_wallet_synced(wallet_node, timeout=30)
     funds = sum(
         calculate_pool_reward(uint32(i)) + calculate_base_farmer_reward(uint32(i)) for i in range(1, num_blocks)
     )
@@ -845,6 +845,11 @@ class OfferSetup:
     maker: StoreSetup
     taker: StoreSetup
     full_node_api: FullNodeSimulator
+    wallet_nodes: list[WalletNode]
+
+    async def wait_for_wallets_synced(self, timeout: int = 30) -> None:
+        for node in self.wallet_nodes:
+            await self.full_node_api.wait_for_wallet_synced(wallet_node=node, timeout=timeout)
 
 
 @pytest.fixture(name="offer_setup")
@@ -857,16 +862,17 @@ async def offer_setup_fixture(
     [full_node_service], wallet_services, bt = two_wallet_nodes_services
     enable_batch_autoinsertion_settings = getattr(request, "param", (True, True))
     full_node_api = full_node_service._api
-    wallets: list[Wallet] = []
+    wallets: list[WalletNode] = []
     for wallet_service in wallet_services:
         wallet_node = wallet_service._node
         assert wallet_node.server is not None
         await wallet_node.server.start_client(PeerInfo(self_hostname, full_node_api.server.get_port()), None)
         assert wallet_node.wallet_state_manager is not None
         wallet = wallet_node.wallet_state_manager.main_wallet
-        wallets.append(wallet)
+        wallets.append(wallet_node)
 
         await full_node_api.farm_blocks_to_wallet(count=1, wallet=wallet, timeout=60)
+        await full_node_api.wait_for_wallet_synced(wallet_node=wallet_node, timeout=30)
 
     async with contextlib.AsyncExitStack() as exit_stack:
         store_setups: list[StoreSetup] = []
@@ -951,6 +957,7 @@ async def offer_setup_fixture(
                 data_rpc_client=taker.data_rpc_client,
             ),
             full_node_api=full_node_api,
+            wallet_nodes=wallets,
         )
 
         maker.data_rpc_client.close()
@@ -1018,6 +1025,7 @@ async def populate_offer_setup(offer_setup: OfferSetup, count: int) -> OfferSetu
             data_rpc_client=offer_setup.taker.data_rpc_client,
         ),
         full_node_api=offer_setup.full_node_api,
+        wallet_nodes=offer_setup.wallet_nodes,
     )
 
 
@@ -1843,6 +1851,7 @@ async def test_make_and_cancel_offer(offer_setup: OfferSetup, reference: MakeAnd
     # due to differences in chain progression, the exact offer and trade id may differ from the reference
     # assert maker_response == {"success": True, "offer": reference.make_offer_response}
     assert maker_response["success"] is True
+    await offer_setup.wait_for_wallets_synced()
 
     cancel_request = {
         "trade_id": maker_response["offer"]["trade_id"],
@@ -1926,6 +1935,7 @@ async def test_make_and_cancel_offer_then_update(
     # due to differences in chain progression, the exact offer and trade id may differ from the reference
     # assert maker_response == {"success": True, "offer": reference.make_offer_response}
     assert maker_response["success"] is True
+    await offer_setup.wait_for_wallets_synced()
 
     cancel_request = {
         "trade_id": maker_response["offer"]["trade_id"],
@@ -2015,6 +2025,7 @@ async def test_make_and_cancel_offer_not_secure_clears_pending_roots(
     # due to differences in chain progression, the exact offer and trade id may differ from the reference
     # assert maker_response == {"success": True, "offer": reference.make_offer_response}
     assert maker_response["success"] is True
+    await offer_setup.wait_for_wallets_synced()
 
     cancel_request = {
         "trade_id": maker_response["offer"]["trade_id"],
@@ -2568,6 +2579,7 @@ async def populate_proof_setup(offer_setup: OfferSetup, count: int) -> OfferSetu
             data_rpc_client=offer_setup.taker.data_rpc_client,
         ),
         full_node_api=offer_setup.full_node_api,
+        wallet_nodes=offer_setup.wallet_nodes,
     )
 
 

chia/_tests/pools/test_pool_rpc.py

@@ -839,6 +839,7 @@ async def test_self_pooling_to_pooling(self, setup: Setup, fee: uint64, self_hos
             our_ph, "", uint32(0), f"{self_hostname}:5000", "new", "SELF_POOLING", fee
         )
         await full_node_api.wait_transaction_records_entered_mempool(records=[creation_tx])
+        await full_node_api.wait_for_wallet_synced(wallet_node=wallet_node, timeout=20)
         creation_tx_2: TransactionRecord = await client.create_new_pool_wallet(
             our_ph, "", uint32(0), f"{self_hostname}:5001", "new", "SELF_POOLING", fee
         )

chia/_tests/wallet/rpc/test_dl_wallet_rpc.py

@@ -77,6 +77,7 @@ async def test_wallet_make_transaction(
             calculate_pool_reward(uint32(i)) + calculate_base_farmer_reward(uint32(i)) for i in range(1, num_blocks)
         )
 
+        await full_node_api.wait_for_wallet_synced(wallet_node)
         await time_out_assert(15, wallet.get_confirmed_balance, initial_funds)
         await time_out_assert(15, wallet.get_unconfirmed_balance, initial_funds)
 
@@ -111,6 +112,7 @@ async def test_wallet_make_transaction(
             for i in range(5):
                 await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(bytes32.zeros))
                 await asyncio.sleep(0.5)
+            await full_node_api.wait_for_wallet_synced(wallet_node)
 
             async def is_singleton_confirmed(rpc_client: WalletRpcClient, lid: bytes32) -> bool:
                 rec = (await rpc_client.dl_latest_singleton(DLLatestSingleton(lid))).singleton
@@ -131,6 +133,7 @@ async def is_singleton_confirmed(rpc_client: WalletRpcClient, lid: bytes32) -> b
             for i in range(5):
                 await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(bytes32.zeros))
                 await asyncio.sleep(0.5)
+            await full_node_api.wait_for_wallet_synced(wallet_node)
 
             new_singleton_record = (await client.dl_latest_singleton(DLLatestSingleton(launcher_id))).singleton
             assert new_singleton_record is not None
@@ -219,13 +222,15 @@ async def is_singleton_generation(rpc_client: WalletRpcClient, lid: bytes32, gen
             launcher_id_2 = (
                 await client.create_new_dl(CreateNewDL(root=merkle_root, fee=uint64(50), push=True), DEFAULT_TX_CONFIG)
             ).launcher_id
+            await full_node_api.wait_for_wallet_synced(wallet_node)
             launcher_id_3 = (
                 await client.create_new_dl(CreateNewDL(root=merkle_root, fee=uint64(50), push=True), DEFAULT_TX_CONFIG)
             ).launcher_id
 
             for i in range(5):
                 await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(bytes32.zeros))
                 await asyncio.sleep(0.5)
+            await full_node_api.wait_for_wallet_synced(wallet_node)
 
             await time_out_assert(15, is_singleton_confirmed, True, client, launcher_id_2)
             await time_out_assert(15, is_singleton_confirmed, True, client, launcher_id_3)
@@ -248,6 +253,7 @@ async def is_singleton_generation(rpc_client: WalletRpcClient, lid: bytes32, gen
             for i in range(5):
                 await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(bytes32.zeros))
                 await asyncio.sleep(0.5)
+            await full_node_api.wait_for_wallet_synced(wallet_node)
 
             await time_out_assert(15, is_singleton_confirmed, True, client, launcher_id)
             await time_out_assert(15, is_singleton_confirmed, True, client, launcher_id_2)
@@ -283,6 +289,7 @@ async def is_singleton_generation(rpc_client: WalletRpcClient, lid: bytes32, gen
             for i in range(5):
                 await full_node_api.farm_new_transaction_block(FarmNewBlockProtocol(bytes32.zeros))
                 await asyncio.sleep(0.5)
+            await full_node_api.wait_for_wallet_synced(wallet_node)
             additions = []
             for tx in txs:
                 if tx.spend_bundle is not None:

chia/_tests/wallet/rpc/test_wallet_rpc.py

@@ -2003,6 +2003,7 @@ async def test_did_endpoints(wallet_rpc_environment: WalletRpcTestEnvironment) -
 
     await time_out_assert(5, check_mempool_spend_count, True, full_node_api, 1)
     await farm_transaction_block(full_node_api, wallet_1_node)
+    await full_node_api.wait_for_wallet_synced(wallet_node=wallet_1_node, timeout=20)
 
     # Update metadata
     with pytest.raises(ValueError, match="wallet id 1 is of type Wallet but type DIDWallet is required"):
@@ -2018,6 +2019,7 @@ async def test_did_endpoints(wallet_rpc_environment: WalletRpcTestEnvironment) -
 
     await time_out_assert(5, check_mempool_spend_count, True, full_node_api, 1)
     await farm_transaction_block(full_node_api, wallet_1_node)
+    await full_node_api.wait_for_wallet_synced(wallet_node=wallet_1_node, timeout=20)
 
     # Transfer DID
     async with wallet_2.wallet_state_manager.new_action_scope(DEFAULT_TX_CONFIG, push=True) as action_scope:
@@ -2031,6 +2033,8 @@ async def test_did_endpoints(wallet_rpc_environment: WalletRpcTestEnvironment) -
 
     await time_out_assert(5, check_mempool_spend_count, True, full_node_api, 1)
     await farm_transaction_block(full_node_api, wallet_1_node)
+    await full_node_api.wait_for_wallet_synced(wallet_node=wallet_1_node, timeout=20)
+    await full_node_api.wait_for_wallet_synced(wallet_node=wallet_2_node, timeout=20)
 
     async def num_wallets() -> int:
         return len(await wallet_2_node.wallet_state_manager.get_all_wallet_info_entries())
@@ -2058,6 +2062,8 @@ async def num_wallets() -> int:
 
     await time_out_assert(5, check_mempool_spend_count, True, full_node_api, 1)
     await farm_transaction_block(full_node_api, wallet_2_node)
+    await full_node_api.wait_for_wallet_synced(wallet_node=wallet_1_node, timeout=20)
+    await full_node_api.wait_for_wallet_synced(wallet_node=wallet_2_node, timeout=20)
 
     next_did_coin = await did_wallet_2.get_coin()
     assert next_did_coin.parent_coin_info == last_did_coin.name()
@@ -2070,6 +2076,8 @@ async def num_wallets() -> int:
 
     await time_out_assert(5, check_mempool_spend_count, True, full_node_api, 1)
     await farm_transaction_block(full_node_api, wallet_2_node)
+    await full_node_api.wait_for_wallet_synced(wallet_node=wallet_1_node, timeout=20)
+    await full_node_api.wait_for_wallet_synced(wallet_node=wallet_2_node, timeout=20)
 
     next_did_coin = await did_wallet_2.get_coin()
     assert next_did_coin.parent_coin_info == last_did_coin.name()

chia/wallet/wallet_rpc_api.py

@@ -322,6 +322,9 @@ def _inner(func: RpcEndpoint) -> RpcEndpoint:
         async def rpc_endpoint(
             self: WalletRpcApi, request: dict[str, Any], *args: object, **kwargs: object
         ) -> EndpointResult:
+            if await self.service.wallet_state_manager.synced() is False:
+                raise ValueError("Wallet needs to be fully synced before making transactions.")
+
             assert self.service.logged_in_fingerprint is not None
             tx_config_loader: TXConfigLoader = TXConfigLoader.from_json_dict(request)
 
@@ -1113,9 +1116,6 @@ async def create_new_wallet(
         extra_conditions: tuple[Condition, ...] = tuple(),
     ) -> EndpointResult:
         wallet_state_manager = self.service.wallet_state_manager
-
-        if await self.service.wallet_state_manager.synced() is False:
-            raise ValueError("Wallet needs to be fully synced.")
         main_wallet = wallet_state_manager.main_wallet
         fee = uint64(request.get("fee", 0))
 
@@ -1631,9 +1631,6 @@ async def send_transaction(
         action_scope: WalletActionScope,
         extra_conditions: tuple[Condition, ...] = tuple(),
     ) -> SendTransactionResponse:
-        if await self.service.wallet_state_manager.synced() is False:
-            raise ValueError("Wallet needs to be fully synced before sending transactions")
-
         wallet = self.service.wallet_state_manager.get_wallet(id=request.wallet_id, required_type=Wallet)
 
         # TODO: Add support for multiple puzhash/amount/memo sets
@@ -2154,8 +2151,6 @@ async def cat_spend(
         extra_conditions: tuple[Condition, ...] = tuple(),
         hold_lock: bool = True,
     ) -> CATSpendResponse:
-        if await self.service.wallet_state_manager.synced() is False:
-            raise ValueError("Wallet needs to be fully synced.")
         wallet = self.service.wallet_state_manager.get_wallet(id=request.wallet_id, required_type=CATWallet)
 
         amounts: list[uint64] = []
@@ -2860,8 +2855,6 @@ async def did_transfer_did(
         action_scope: WalletActionScope,
         extra_conditions: tuple[Condition, ...] = tuple(),
     ) -> DIDTransferDIDResponse:
-        if await self.service.wallet_state_manager.synced() is False:
-            raise ValueError("Wallet needs to be fully synced.")
         did_wallet = self.service.wallet_state_manager.get_wallet(id=request.wallet_id, required_type=DIDWallet)
         puzzle_hash: bytes32 = decode_puzzle_hash(request.inner_address)
         async with self.service.wallet_state_manager.lock:
@@ -3331,8 +3324,6 @@ async def nft_mint_bulk(
     ) -> NFTMintBulkResponse:
         if action_scope.config.push:
             raise ValueError("Automatic pushing of nft minting transactions not yet available")  # pragma: no cover
-        if await self.service.wallet_state_manager.synced() is False:
-            raise ValueError("Wallet needs to be fully synced.")
         nft_wallet = self.service.wallet_state_manager.get_wallet(id=request.wallet_id, required_type=NFTWallet)
         if request.royalty_address in {None, ""}:
             royalty_puzhash = await action_scope.get_puzzle_hash(self.service.wallet_state_manager)
@@ -3610,9 +3601,6 @@ async def pw_join_pool(
     ) -> PWJoinPoolResponse:
         wallet = self.service.wallet_state_manager.get_wallet(id=request.wallet_id, required_type=PoolWallet)
 
-        if await self.service.wallet_state_manager.synced() is False:
-            raise ValueError("Wallet needs to be fully synced.")
-
         pool_wallet_info: PoolWalletInfo = await wallet.get_current_state()
         if (
             pool_wallet_info.current.state == FARMING_TO_POOL.value
@@ -3652,9 +3640,6 @@ async def pw_self_pool(
         # Then we transition to FARMING_TO_POOL or SELF_POOLING
         wallet = self.service.wallet_state_manager.get_wallet(id=request.wallet_id, required_type=PoolWallet)
 
-        if await self.service.wallet_state_manager.synced() is False:
-            raise ValueError("Wallet needs to be fully synced.")
-
         total_fee = await wallet.self_pool(request.fee, action_scope)
         # tx_endpoint will take care of filling in these default values
         return PWSelfPoolResponse(
@@ -3674,8 +3659,6 @@ async def pw_absorb_rewards(
         extra_conditions: tuple[Condition, ...] = tuple(),
     ) -> PWAbsorbRewardsResponse:
         """Perform a sweep of the p2_singleton rewards controlled by the pool wallet singleton"""
-        if await self.service.wallet_state_manager.synced() is False:
-            raise ValueError("Wallet needs to be fully synced before collecting rewards")
         wallet = self.service.wallet_state_manager.get_wallet(id=request.wallet_id, required_type=PoolWallet)
 
         assert isinstance(wallet, PoolWallet)