Improve requesting transactions advertised via NewTransaction.

Author: AmineKhaldi

chia/_tests/core/full_node/test_full_node.py

@@ -1169,6 +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.respond_transaction_allowance = 1
     res = await full_node_1.respond_transaction(respond_transaction, peer)
     assert res is None
 
@@ -3778,3 +3779,36 @@ 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 disconnected
+    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)
+    # A peer with allowance x can send x messages without being disconnected
+    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
+    await wsc2.send_message(msg)
+    await asyncio.sleep(1)
+    assert wsc2.closed is False
+    # Now the x+1 message gets it disconnected
+    await wsc2.send_message(msg)
+    await time_out_assert(5, lambda: wsc2.closed)

chia/_tests/wallet/simple_sync/test_simple_sync_protocol.py

@@ -506,6 +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.respond_transaction_allowance = 1
     await full_node_api.respond_transaction(RespondTransaction(tx), fake_wallet_peer)
 
     await full_node_api.process_spend_bundles(bundles=[tx])
@@ -557,6 +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.respond_transaction_allowance = 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
@@ -615,6 +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.respond_transaction_allowance = 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

@@ -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,29 @@ 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 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 +331,15 @@ 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"Disconnecting peer {peer.get_peer_logging()} for sending unsolicited RespondTransaction"
+                )
+                await peer.close()
+                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)