Improve requesting transactions advertised via NewTransaction.

Author: AmineKhaldi

chia/_tests/core/full_node/test_full_node.py

@@ -990,9 +990,6 @@ async def test_new_transaction_and_mempool(
     await full_node_1.new_transaction(new_transaction, fake_peer)
     await time_out_assert(10, new_transaction_requested, True, incoming_queue, new_transaction)
 
-    respond_transaction_2 = fnp.RespondTransaction(spend_bundle)
-    await full_node_1.respond_transaction(respond_transaction_2, peer)
-
     blocks = bt.get_consecutive_blocks(
         1,
         block_list_input=blocks,
@@ -1039,9 +1036,8 @@ async def test_new_transaction_and_mempool(
         ]
         spend_bundle = WalletSpendBundle.aggregate(spend_bundles)
         assert estimate_fees(spend_bundle) == fee
-        respond_transaction = wallet_protocol.SendTransaction(spend_bundle)
 
-        await full_node_1.send_transaction(respond_transaction, fake_peer)
+        await full_node_1.send_transaction(SendTransaction(spend_bundle), fake_peer)
 
         request = fnp.RequestTransaction(spend_bundle.get_hash())
         req = await full_node_1.request_transaction(request)
@@ -1168,9 +1164,8 @@ async def test_request_respond_transaction(
     coin = find_reward_coin(blocks[-1], wallet_ph)
     spend_bundle = wallet_a.generate_signed_transaction(uint64(100), receiver_puzzlehash, coin)
     assert spend_bundle is not None
-    respond_transaction = fnp.RespondTransaction(spend_bundle)
-    res = await full_node_1.respond_transaction(respond_transaction, peer)
-    assert res is None
+    res = await full_node_1.send_transaction(SendTransaction(spend_bundle), peer)
+    assert res is not None
 
     # Check broadcast
     await time_out_assert(10, time_out_messages(incoming_queue, "new_transaction"))
@@ -1182,7 +1177,7 @@ async def test_request_respond_transaction(
 
 
 @pytest.mark.anyio
-async def test_respond_transaction_fail(
+async def test_send_transaction_fail(
     wallet_nodes: tuple[
         FullNodeSimulator, FullNodeSimulator, ChiaServer, ChiaServer, WalletTool, WalletTool, BlockTools
     ],
@@ -1223,9 +1218,11 @@ async def test_respond_transaction_fail(
     spend_bundle = wallet_a.generate_signed_transaction(uint64(100_000_000_000_000), receiver_puzzlehash, coin)
 
     assert spend_bundle is not None
-    respond_transaction = fnp.RespondTransaction(spend_bundle)
-    msg = await full_node_1.respond_transaction(respond_transaction, peer)
-    assert msg is None
+    response_msg = await full_node_1.send_transaction(SendTransaction(spend_bundle), peer)
+    assert response_msg is not None
+    assert TransactionAck.from_bytes(response_msg.data) == TransactionAck(
+        txid=spend_bundle.name(), status=uint8(MempoolInclusionStatus.FAILED), error="MINTING_COIN"
+    )
 
     await asyncio.sleep(1)
     assert incoming_queue.qsize() == 0
@@ -3778,3 +3775,40 @@ async def test_register_for_coin_updates(
     assert {cs.coin.name() for cs in response_data.coin_states} == set(first_coin)
     for coin_id in remaining_coins:
         assert coin_id not in response_data.coin_ids
+
+
+@pytest.mark.anyio
+@pytest.mark.limit_consensus_modes(allowed=[ConsensusMode.HARD_FORK_2_0], reason="irrelevant")
+async def test_respond_transaction(
+    one_node_one_block: tuple[FullNodeSimulator, ChiaServer, BlockTools], self_hostname: str
+) -> None:
+    _, server, _ = one_node_one_block
+    sb = SpendBundle([], G2Element())
+    msg = make_msg(ProtocolMessageTypes.respond_transaction, bytes(fnp.RespondTransaction(sb)))
+    # A peer with no allowance should get banned
+    wsc, peer_id = await add_dummy_connection_wsc(server, self_hostname, 42)
+    await time_out_assert(5, lambda: peer_id in server.all_connections)
+    conn = server.all_connections[peer_id]
+    conn.peer_info = PeerInfo("1.3.3.7", conn.peer_info.port)
+    assert conn.respond_transaction_allowance == 0
+    await wsc.send_message(msg)
+    await time_out_assert(5, lambda: wsc.closed)
+    await time_out_assert(5, lambda: "1.3.3.7" in server.banned_peers)
+    # A peer with allowance x can send x messages without being banned
+    wsc2, peer_id2 = await add_dummy_connection_wsc(server, self_hostname, 42)
+    await time_out_assert(5, lambda: peer_id2 in server.all_connections)
+    conn2 = server.all_connections[peer_id2]
+    conn2.peer_info = PeerInfo("1.2.3.4", conn2.peer_info.port)
+    conn2.respond_transaction_allowance = 2
+    await wsc2.send_message(msg)
+    await asyncio.sleep(1)
+    assert wsc2.closed is False
+    assert "1.2.3.4" not in server.banned_peers
+    await wsc2.send_message(msg)
+    await asyncio.sleep(1)
+    assert wsc2.closed is False
+    assert "1.2.3.4" not in server.banned_peers
+    # Now the x+1 message gets it banned
+    await wsc2.send_message(msg)
+    await time_out_assert(5, lambda: wsc2.closed)
+    await time_out_assert(5, lambda: "1.2.3.4" in server.banned_peers)

chia/_tests/core/full_node/test_performance.py

@@ -73,8 +73,7 @@ async def test_full_block_performance(
             )
             assert spend_bundle is not None
 
-            respond_transaction_2 = fnp.RespondTransaction(spend_bundle)
-            await full_node_1.respond_transaction(respond_transaction_2, fake_peer)
+            await full_node_1.full_node.add_transaction(spend_bundle, spend_bundle.name(), fake_peer)
 
             blocks = bt.get_consecutive_blocks(
                 1,
@@ -107,9 +106,7 @@ async def test_full_block_performance(
             num_tx: int = 0
             for spend_bundle, spend_bundle_id in zip(spend_bundles, spend_bundle_ids):
                 num_tx += 1
-                respond_transaction = fnp.RespondTransaction(spend_bundle)
-
-                await full_node_1.respond_transaction(respond_transaction, fake_peer)
+                await full_node_1.full_node.add_transaction(spend_bundle, spend_bundle_id, fake_peer)
 
                 request_transaction = fnp.RequestTransaction(spend_bundle_id)
                 req = await full_node_1.request_transaction(request_transaction)

chia/_tests/core/mempool/test_mempool.py

@@ -425,8 +425,7 @@ async def test_basic_mempool_manager(
         coin = await next_block(full_node_1, wallet_a, bt)
         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)
-        await full_node_1.respond_transaction(tx, peer, test=True)
+        await full_node_1.send_transaction(wallet_protocol.SendTransaction(spend_bundle), peer, test=True)
 
         await time_out_assert(
             10,

chia/_tests/wallet/simple_sync/test_simple_sync_protocol.py

@@ -13,7 +13,6 @@
 from chia._tests.util.time_out_assert import time_out_assert
 from chia.consensus.block_rewards import calculate_base_farmer_reward, calculate_pool_reward
 from chia.protocols import wallet_protocol
-from chia.protocols.full_node_protocol import RespondTransaction
 from chia.protocols.outbound_message import Message, NodeType
 from chia.protocols.protocol_message_types import ProtocolMessageTypes
 from chia.protocols.wallet_protocol import CoinStateUpdate, RespondToCoinUpdates
@@ -506,7 +505,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)
-    await full_node_api.respond_transaction(RespondTransaction(tx), fake_wallet_peer)
+    await full_node_api.send_transaction(wallet_protocol.SendTransaction(tx), fake_wallet_peer)
 
     await full_node_api.process_spend_bundles(bundles=[tx])
 
@@ -557,7 +556,7 @@ async def test_subscribe_for_puzzle_hash_coin_hint_duplicates(
             ConditionOpcode.CREATE_COIN: [ConditionWithArgs(ConditionOpcode.CREATE_COIN, [ph, int_to_bytes(1), ph])]
         },
     )
-    await full_node_api.respond_transaction(RespondTransaction(tx), wallet_connection)
+    await full_node_api.send_transaction(wallet_protocol.SendTransaction(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
     msg = wallet_protocol.RegisterForPhUpdates([ph], uint32(0))
@@ -615,7 +614,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)
-    await full_node_api.respond_transaction(RespondTransaction(tx), fake_wallet_peer)
+    await full_node_api.send_transaction(wallet_protocol.SendTransaction(tx), fake_wallet_peer)
 
     await full_node_api.process_spend_bundles(bundles=[tx])
 

chia/full_node/full_node.py

@@ -939,7 +939,7 @@ async def on_connect(self, connection: WSChiaConnection) -> None:
             if synced and peak_height is not None:
                 my_filter = self.mempool_manager.get_filter()
                 mempool_request = full_node_protocol.RequestMempoolTransactions(my_filter)
-
+                connection.respond_transaction_allowance = 100
                 msg = make_msg(ProtocolMessageTypes.request_mempool_transactions, mempool_request)
                 await connection.send_message(msg)
 

chia/full_node/full_node_api.py

@@ -110,12 +110,26 @@ 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
+            response = await random_peer.call_api(
+                FullNodeAPI.request_transaction, full_node_protocol.RequestTransaction(transaction_id), timeout=5
+            )
+            if response is None or not isinstance(response, full_node_protocol.RespondTransaction):
+                continue
+            try:
+                full_node.transaction_queue.put(
+                    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,
+                    ),
+                    random_peer.peer_node_id,
+                )
+            except TransactionQueueFull:
+                continue
+            break
     except asyncio.CancelledError:
         pass
     finally:
@@ -314,6 +328,13 @@ async def respond_transaction(
         spend_name = std_hash(tx_bytes)
         if spend_name in self.full_node.full_node_store.pending_tx_request:
             self.full_node.full_node_store.pending_tx_request.pop(spend_name)
+        else:
+            if not test and peer.respond_transaction_allowance <= 0:
+                self.log.warning(f"Banning peer {peer.get_peer_logging()} for sending unsolicited RespondTransaction")
+                await peer.close(CONSENSUS_ERROR_BAN_SECONDS)
+                return None
+            peer.respond_transaction_allowance -= 1
+
         peers_with_tx = {}
         if spend_name in self.full_node.full_node_store.peers_with_tx:
             peers_with_tx = self.full_node.full_node_store.peers_with_tx.pop(spend_name)

chia/server/ws_connection.py

@@ -123,6 +123,7 @@ class WSChiaConnection:
     closed: bool = False
     connection_type: NodeType | None = None
     request_nonce: uint16 = uint16(0)
+    respond_transaction_allowance: int = 0
     peer_capabilities: list[Capability] = field(default_factory=list)
     # Used by the Chia Seeder.
     version: str = field(default_factory=str)