refactor: move rate limit to rpc class

Author: LouisTsai-Csie

packages/testing/src/execution_testing/cli/pytest_commands/plugins/execute/rpc/chain_builder_eth_rpc.py

@@ -205,11 +205,13 @@ def __init__(
         get_payload_wait_time: float,
         initial_forkchoice_update_retries: int = 5,
         transaction_wait_timeout: int = 60,
+        max_transactions_per_batch: int | None = None,
     ):
         """Initialize the Ethereum RPC client for the hive simulator."""
         super().__init__(
             rpc_endpoint,
             transaction_wait_timeout=transaction_wait_timeout,
+            max_transactions_per_batch=max_transactions_per_batch,
         )
         self.fork = fork
         self.engine_rpc = engine_rpc

packages/testing/src/execution_testing/cli/pytest_commands/plugins/execute/rpc/hive.py

@@ -383,6 +383,7 @@ def eth_rpc(
     session_fork: Fork,
     transactions_per_block: int,
     session_temp_folder: Path,
+    max_transactions_per_batch: int | None,
 ) -> EthRPC:
     """Initialize ethereum RPC client for the execution client under test."""
     get_payload_wait_time = request.config.getoption("get_payload_wait_time")
@@ -395,4 +396,5 @@ def eth_rpc(
         session_temp_folder=session_temp_folder,
         get_payload_wait_time=get_payload_wait_time,
         transaction_wait_timeout=tx_wait_timeout,
+        max_transactions_per_batch=max_transactions_per_batch,
     )

packages/testing/src/execution_testing/cli/pytest_commands/plugins/execute/rpc/remote.py

@@ -149,11 +149,16 @@ def eth_rpc(
     session_fork: Fork,
     transactions_per_block: int,
     session_temp_folder: Path,
+    max_transactions_per_batch: int | None,
 ) -> EthRPC:
     """Initialize ethereum RPC client for the execution client under test."""
     tx_wait_timeout = request.config.getoption("tx_wait_timeout")
     if engine_rpc is None:
-        return EthRPC(rpc_endpoint, transaction_wait_timeout=tx_wait_timeout)
+        return EthRPC(
+            rpc_endpoint,
+            transaction_wait_timeout=tx_wait_timeout,
+            max_transactions_per_batch=max_transactions_per_batch,
+        )
     get_payload_wait_time = request.config.getoption("get_payload_wait_time")
     return ChainBuilderEthRPC(
         rpc_endpoint=rpc_endpoint,
@@ -163,4 +168,5 @@ def eth_rpc(
         session_temp_folder=session_temp_folder,
         get_payload_wait_time=get_payload_wait_time,
         transaction_wait_timeout=tx_wait_timeout,
+        max_transactions_per_batch=max_transactions_per_batch,
     )

packages/testing/src/execution_testing/execution/transaction_post.py

@@ -23,9 +23,6 @@
 
 logger = get_logger(__name__)
 
-# Threshold above which RPC service may become unstable
-RPC_OVERLOAD_THRESHOLD = 1000
-
 
 class TransactionPost(BaseExecute):
     """
@@ -41,8 +38,6 @@ class TransactionPost(BaseExecute):
     skip_gas_used_validation: bool = (
         False  # Skip gas validation even if expected is set
     )
-    # Transaction batching to avoid RPC overload
-    max_transactions_per_batch: int = 750
 
     format_name: ClassVar[str] = "transaction_post_test"
     description: ClassVar[str] = (
@@ -86,21 +81,6 @@ def execute(
         del fork
         del engine_rpc
 
-        # Check if CLI override for max_transactions_per_batch is provided
-        cli_max_tx_per_batch = request.config.getoption(
-            "max_tx_per_batch", None
-        )
-        if cli_max_tx_per_batch is not None:
-            self.max_transactions_per_batch = cli_max_tx_per_batch
-
-        # Warn if max_transactions_per_batch exceeds safe threshold
-        if self.max_transactions_per_batch > RPC_OVERLOAD_THRESHOLD:
-            logger.warning(
-                f"max_transactions_per_batch ({self.max_transactions_per_batch}) exceeds "
-                f"the safe threshold ({RPC_OVERLOAD_THRESHOLD}). "
-                "This may cause RPC service instability or failures."
-            )
-
         for block in self.blocks:
             for tx in block:
                 if not isinstance(tx, NetworkWrappedTransaction):
@@ -152,18 +132,9 @@ def execute(
                             f"Transaction rejected as expected: {exc_info.value}"
                         )
             else:
-                # Send transactions in batches to avoid RPC overload
-                batch_size = self.max_transactions_per_batch
-                for i in range(0, len(signed_txs), batch_size):
-                    batch = signed_txs[i : i + batch_size]
-                    logger.info(
-                        f"Sending transaction batch {i // batch_size + 1} "
-                        f"({len(batch)} transactions, "
-                        f"{i + 1}-{min(i + batch_size, len(signed_txs))} "
-                        f"of {len(signed_txs)})"
-                    )
-                    eth_rpc.send_wait_transactions(batch)
-                    all_tx_hashes.extend([tx.hash for tx in batch])
+                # Send transactions (batching is handled by eth_rpc internally)
+                eth_rpc.send_wait_transactions(signed_txs)
+                all_tx_hashes.extend([tx.hash for tx in signed_txs])
 
         # Perform gas validation if required for benchmarking
         # Ensures benchmark tests consume exactly the expected gas

packages/testing/src/execution_testing/rpc/rpc.py

@@ -198,8 +198,12 @@ class EthRPC(BaseRPC):
     within EEST based hive simulators.
     """
 
+    OVERLOAD_THRESHOLD: int = 1000
+    DEFAULT_MAX_TRANSACTIONS_PER_BATCH: int = 750
+
     transaction_wait_timeout: int = 60
     poll_interval: float = 1.0  # how often to poll for tx inclusion
+    max_transactions_per_batch: int = DEFAULT_MAX_TRANSACTIONS_PER_BATCH
 
     gas_information_stale_seconds: int
 
@@ -214,6 +218,7 @@ def __init__(
         transaction_wait_timeout: int = 60,
         poll_interval: float | None = None,
         gas_information_stale_seconds: int = 12,
+        max_transactions_per_batch: int | None = None,
         **kwargs: Any,
     ) -> None:
         """
@@ -251,6 +256,17 @@ def __init__(
             "blobBaseFee": 0.0,
         }
 
+        # Transaction batching configuration
+        if max_transactions_per_batch is None:
+            max_transactions_per_batch = self.DEFAULT_MAX_TRANSACTIONS_PER_BATCH
+        self.max_transactions_per_batch = max_transactions_per_batch
+        if max_transactions_per_batch > self.OVERLOAD_THRESHOLD:
+            logger.warning(
+                f"max_transactions_per_batch ({max_transactions_per_batch}) exceeds "
+                f"the safe threshold ({self.OVERLOAD_THRESHOLD}). "
+                "This may cause RPC service instability or failures."
+            )
+
     def config(self, timeout: int | None = None) -> EthConfigResponse | None:
         """
         `eth_config`: Returns information about a fork configuration of the
@@ -581,10 +597,25 @@ def send_wait_transactions(
     ) -> List[Any]:
         """
         Send list of transactions and waits until all of them are included in a
-        block.
-        """
-        self.send_transactions(transactions)
-        return self.wait_for_transactions(transactions)
+        block. Transactions are sent in batches to avoid RPC overload.
+        """
+        results: List[Any] = []
+        batch_size = self.max_transactions_per_batch
+        total_txs = len(transactions)
+
+        for i in range(0, total_txs, batch_size):
+            batch = transactions[i : i + batch_size]
+            if total_txs > batch_size:
+                logger.info(
+                    f"Sending transaction batch {i // batch_size + 1} "
+                    f"({len(batch)} transactions, "
+                    f"{i + 1}-{min(i + batch_size, total_txs)} "
+                    f"of {total_txs})"
+                )
+            self.send_transactions(batch)
+            results.extend(self.wait_for_transactions(batch))
+
+        return results
 
 
 class DebugRPC(EthRPC):