Improve requesting transactions advertised via NewTransaction.

Author: AmineKhaldi

chia/_tests/core/full_node/test_full_node.py

@@ -1169,7 +1169,7 @@ async def test_request_respond_transaction(
     spend_bundle = wallet_a.generate_signed_transaction(uint64(100), receiver_puzzlehash, coin)
     assert spend_bundle is not None
     respond_transaction = fnp.RespondTransaction(spend_bundle)
-    peer.expected_mempool_responses += 1
+    peer.expected_mempool_responses = 1
     res = await full_node_1.respond_transaction(respond_transaction, peer)
     assert res is None
 
@@ -1225,7 +1225,7 @@ async def test_respond_transaction_fail(
 
     assert spend_bundle is not None
     respond_transaction = fnp.RespondTransaction(spend_bundle)
-    peer.expected_mempool_responses += 1
+    peer.expected_mempool_responses = 1
     msg = await full_node_1.respond_transaction(respond_transaction, peer)
     assert msg is None
 

chia/_tests/core/mempool/test_mempool.py

@@ -465,7 +465,7 @@ async def test_basic_mempool_manager(
         spend_bundle = generate_test_spend_bundle(wallet_a, coin)
         assert spend_bundle is not None
         tx: full_node_protocol.RespondTransaction = full_node_protocol.RespondTransaction(spend_bundle)
-        peer.expected_mempool_responses += 1
+        peer.expected_mempool_responses = 1
         await full_node_1.respond_transaction(tx, peer, test=True)
 
         await time_out_assert(

chia/_tests/wallet/simple_sync/test_simple_sync_protocol.py

@@ -506,7 +506,7 @@ async def test_subscribe_for_hint(simulator_and_wallet: OldSimulatorsAndWallets,
     await full_node_api.wait_for_wallet_synced(wallet_node=wallet_node, timeout=20)
 
     tx = wt.generate_signed_transaction(uint64(10), wt.get_new_puzzlehash(), coin_spent, condition_dic=condition_dict)
-    fake_wallet_peer.expected_mempool_responses += 1
+    fake_wallet_peer.expected_mempool_responses = 1
     await full_node_api.respond_transaction(RespondTransaction(tx), fake_wallet_peer)
 
     await full_node_api.process_spend_bundles(bundles=[tx])
@@ -558,7 +558,7 @@ async def test_subscribe_for_puzzle_hash_coin_hint_duplicates(
             ConditionOpcode.CREATE_COIN: [ConditionWithArgs(ConditionOpcode.CREATE_COIN, [ph, int_to_bytes(1), ph])]
         },
     )
-    wallet_connection.expected_mempool_responses += 1
+    wallet_connection.expected_mempool_responses = 1
     await full_node_api.respond_transaction(RespondTransaction(tx), wallet_connection)
     await full_node_api.process_spend_bundles(bundles=[tx])
     # Query the coin states and make sure it doesn't contain duplicated entries
@@ -617,7 +617,7 @@ async def test_subscribe_for_hint_long_sync(
     await full_node_api.wait_for_wallet_synced(wallet_node=wallet_node, timeout=20)
 
     tx = wt.generate_signed_transaction(uint64(10), wt.get_new_puzzlehash(), coin_spent, condition_dic=condition_dict)
-    fake_wallet_peer.expected_mempool_responses += 1
+    fake_wallet_peer.expected_mempool_responses = 1
     await full_node_api.respond_transaction(RespondTransaction(tx), fake_wallet_peer)
 
     await full_node_api.process_spend_bundles(bundles=[tx])

chia/full_node/full_node.py

@@ -949,7 +949,7 @@ async def on_connect(self, connection: WSChiaConnection) -> None:
                 except Exception:
                     old_peer = True
                 if old_peer:
-                    connection.expected_mempool_responses += 100
+                    connection.expected_mempool_responses = 100
 
         peak_full: FullBlock | None = await self.blockchain.get_full_peak()
 

chia/full_node/full_node_api.py

@@ -110,12 +110,32 @@ async def tx_request_and_timeout(full_node: FullNode, transaction_id: bytes32, t
             if peer_id not in full_node.server.all_connections:
                 continue
             random_peer = full_node.server.all_connections[peer_id]
-            request_tx = full_node_protocol.RequestTransaction(transaction_id)
-            msg = make_msg(ProtocolMessageTypes.request_transaction, request_tx)
-            await random_peer.send_message(msg)
-            await asyncio.sleep(5)
-            if full_node.mempool_manager.seen(transaction_id):
-                break
+            try:
+                response = await random_peer.call_api(
+                    FullNodeAPI.request_transaction, full_node_protocol.RequestTransaction(transaction_id), timeout=5
+                )
+            except Exception:
+                continue
+            if not isinstance(response, full_node_protocol.RespondTransaction):
+                continue
+            entry = TransactionQueueEntry(
+                transaction=response.transaction,
+                transaction_bytes=bytes(response.transaction),
+                spend_name=response.transaction.name(),
+                peer=random_peer,
+                test=False,
+                peers_with_tx=peers_with_tx,
+            )
+            # Try to put the transaction in this peer's queue, but if it's full
+            # then try another peer's queue. `TransactionQueue` has an internal
+            # queue for each peer.
+            for pid in [random_peer.peer_node_id, *peers_with_tx]:
+                try:
+                    full_node.transaction_queue.put(entry, pid)
+                    break
+                except TransactionQueueFull:
+                    continue
+            break
     except asyncio.CancelledError:
         pass
     finally: