Support for JSON RPC simulateTransaction method

scripts/install_poetry.sh

@@ -1,3 +1,3 @@
 set -eu
 
-pip3 install -U poetry==1.3
+pip3 install -U poetry==1.3.1

starknet_devnet/blueprints/feeder_gateway.py

@@ -374,7 +374,7 @@ async def estimate_fee_bulk():
     block_id = _get_block_id(request.args)
     skip_validate = _get_skip_validate(request.args)
 
-    _, fee_responses = await state.starknet_wrapper.calculate_traces_and_fees(
+    _, fee_responses, _ = await state.starknet_wrapper.calculate_traces_and_fees(
         transactions,
         block_id=block_id,
         skip_validate=skip_validate,

starknet_devnet/blueprints/rpc/routes.py

@@ -45,6 +45,7 @@
     get_transaction_by_hash,
     get_transaction_receipt,
     pending_transactions,
+    simulate_transaction,
 )
 from starknet_devnet.blueprints.rpc.utils import rpc_error, rpc_response
 from starknet_devnet.util import StarknetDevnetException
@@ -73,6 +74,7 @@
     "addInvokeTransaction": add_invoke_transaction,
     "addDeclareTransaction": add_declare_transaction,
     "addDeployAccountTransaction": add_deploy_account_transaction,
+    "simulateTransaction": simulate_transaction,
 }
 
 rpc = Blueprint("rpc", __name__, url_prefix="/rpc")

starknet_devnet/blueprints/rpc/schema.py

@@ -14,6 +14,7 @@
 
 from starknet_devnet.blueprints.rpc.rpc_spec import RPC_SPECIFICATION
 from starknet_devnet.blueprints.rpc.rpc_spec_write import RPC_SPECIFICATION_WRITE
+from starknet_devnet.blueprints.rpc.rpc_trace_spec import RPC_SPECIFICATION_TRACE
 from starknet_devnet.state import state
 
 
@@ -22,9 +23,14 @@
 def _load_schemas() -> Tuple[Dict[str, Any], Dict[str, Any]]:
     specs_json = json.loads(RPC_SPECIFICATION)
     write_specs_json = json.loads(RPC_SPECIFICATION_WRITE)
+    trace_specs_json = json.loads(RPC_SPECIFICATION_TRACE)
     schemas = specs_json["components"]["schemas"]
 
-    methods = {**_extract_methods(specs_json), **_extract_methods(write_specs_json)}
+    methods = {
+        **_extract_methods(specs_json),
+        **_extract_methods(write_specs_json),
+        **_extract_methods(trace_specs_json),
+    }
 
     return methods, schemas
 

starknet_devnet/blueprints/rpc/structures/payloads.py

@@ -1,9 +1,12 @@
+# pylint: disable=too-many-lines
+
 """
 RPC payload structures
 """
 
 from __future__ import annotations
 
+from enum import Enum, auto
 from typing import Callable, Dict, List, Optional, Union
 
 from marshmallow.exceptions import MarshmallowError
@@ -27,6 +30,7 @@
     L1HandlerSpecificInfo,
     StarknetBlock,
     TransactionSpecificInfo,
+    TransactionTrace,
     TransactionType,
 )
 from starkware.starknet.services.api.gateway.transaction import (
@@ -619,6 +623,13 @@ class StructAbiEntry(TypedDict):
 AbiEntry = Union[FunctionAbiEntry, EventAbiEntry, StructAbiEntry]
 
 
+class SimulationFlag(Enum):
+    """Enum with flags for simulate transaction"""
+
+    SKIP_VALIDATE = auto()
+    SKIP_EXECUTE = auto()
+
+
 def function_abi_entry(abi_entry: AbiEntryType) -> FunctionAbiEntry:
     """
     Convert function gateway abi entry to rpc FunctionAbiEntry
@@ -944,3 +955,71 @@ def nonces() -> List[RpcNonceDiff]:
         "state_diff": state_diff,
     }
     return rpc_state
+
+
+def rpc_map_traces(
+    traces: List[TransactionTrace], types: List[TransactionType]
+) -> list:
+    """
+    Mapping for traces based on transaction types.
+    """
+    # traces number must be equal to types to properly map objects
+    assert len(traces) == len(types)
+
+    result = []
+    for i, trace in enumerate(traces):
+        trace_dict = trace.dump()
+
+        if types[i] == TransactionType.INVOKE_FUNCTION:
+            trace = rpc_invoke_txn_trace(trace_dict)
+        elif types[i] == TransactionType.DECLARE:
+            trace = rpc_declare_txn_trace(trace_dict)
+        elif types[i] == TransactionType.DEPLOY_ACCOUNT:
+            trace = rpc_deploy_account_txn_trace(trace_dict)
+        elif types[i] == TransactionType.L1_HANDLER:
+            trace = rpc_l1_handler_txn_trace(trace_dict)
+
+        result.append(trace)
+
+    return result
+
+
+def rpc_invoke_txn_trace(trace_dict: dict):
+    """
+    Mapping for the execution trace of an invoke transaction.
+    """
+    return {
+        "validate_invocation": trace_dict.get("validate_invocation", None),
+        "execute_invocation": trace_dict.get("function_invocation", None),
+        "fee_transfer_invocation": trace_dict.get("fee_transfer_invocation", None),
+    }
+
+
+def rpc_declare_txn_trace(trace_dict: dict):
+    """
+    Mapping for the execution trace of an declare transaction.
+    """
+    return {
+        "validate_invocation": trace_dict.get("validate_invocation", None),
+        "fee_transfer_invocation": trace_dict.get("fee_transfer_invocation", None),
+    }
+
+
+def rpc_deploy_account_txn_trace(trace_dict: dict):
+    """
+    Mapping for the execution trace of an deploy account transaction.
+    """
+    return {
+        "validate_invocation": trace_dict.get("validate_invocation", None),
+        "constructor_invocation": trace_dict.get("function_invocation", None),
+        "fee_transfer_invocation": trace_dict.get("fee_transfer_invocation", None),
+    }
+
+
+def rpc_l1_handler_txn_trace(trace_dict: dict):
+    """
+    Mapping for the execution trace of an l1 handler transaction.
+    """
+    return {
+        "function_invocation": trace_dict.get("validate_invocation", None),
+    }

starknet_devnet/blueprints/rpc/transactions.py

@@ -17,10 +17,12 @@
     RpcBroadcastedInvokeTxn,
     RpcBroadcastedTxn,
     RpcTransaction,
+    SimulationFlag,
     make_declare,
     make_deploy_account,
     make_invoke_function,
     rpc_fee_estimate,
+    rpc_map_traces,
     rpc_transaction,
 )
 from starknet_devnet.blueprints.rpc.structures.responses import (
@@ -198,7 +200,7 @@ async def estimate_fee(request: List[RpcBroadcastedTxn], block_id: BlockId) -> l
     transactions = list(map(make_transaction, request))
 
     try:
-        _, fee_response = await state.starknet_wrapper.calculate_traces_and_fees(
+        _, fee_response, _ = await state.starknet_wrapper.calculate_traces_and_fees(
             transactions,
             skip_validate=False,
             block_id=block_id,
@@ -213,3 +215,46 @@ async def estimate_fee(request: List[RpcBroadcastedTxn], block_id: BlockId) -> l
         raise RpcError(code=-1, message=ex.message) from ex
 
     return rpc_fee_estimate(fee_response)
+
+
+@validate_schema("simulateTransaction")
+async def simulate_transaction(
+    block_id: BlockId,
+    transaction: List[RpcTransaction],
+    simulation_flags: List[SimulationFlag],
+) -> list:
+    """
+    Simulate transactions.
+    SKIP_EXECUTE SimulationFlag is not supported.
+    """
+    await assert_block_id_is_valid(block_id)
+    transactions = list(map(make_transaction, transaction))
+    skip_validate = SimulationFlag.SKIP_VALIDATE.name in simulation_flags
+    simulated_transactions = []
+
+    try:
+        (
+            traces,
+            fee,
+            transaction_types,
+        ) = await state.starknet_wrapper.calculate_traces_and_fees(
+            transactions,
+            skip_validate=skip_validate,
+            block_id=block_id,
+        )
+        simulated_transactions.append(
+            {
+                "transaction_trace": rpc_map_traces(traces, transaction_types),
+                "fee_estimation": rpc_fee_estimate(fee),
+            }
+        )
+    except StarkException as ex:
+        if "Entry point" in ex.message and "not found" in ex.message:
+            raise RpcError.from_spec_name("INVALID_MESSAGE_SELECTOR") from ex
+        if "While handling calldata" in ex.message:
+            raise RpcError.from_spec_name("INVALID_CALL_DATA") from ex
+        if "is not deployed" in ex.message:
+            raise RpcError.from_spec_name("CONTRACT_NOT_FOUND") from ex
+        raise RpcError(code=-1, message=ex.message) from ex
+
+    return simulated_transactions

starknet_devnet/starknet_wrapper.py

@@ -828,7 +828,7 @@ async def calculate_trace_and_fee(
         block_id: BlockId = DEFAULT_BLOCK_ID,
     ):
         """Calculates trace and fee by simulating tx on state copy."""
-        traces, fees = await self.calculate_traces_and_fees(
+        traces, fees, _ = await self.calculate_traces_and_fees(
             [external_tx], skip_validate=skip_validate, block_id=block_id
         )
         assert len(traces) == len(fees) == 1
@@ -847,6 +847,7 @@ async def calculate_traces_and_fees(
 
         traces = []
         fee_estimation_infos = []
+        transaction_types = []
 
         for external_tx in external_txs:
             # pylint: disable=protected-access
@@ -888,8 +889,10 @@ async def calculate_traces_and_fees(
             )
             fee_estimation_infos.append(fee_estimation_info)
 
+            transaction_types.append(internal_tx.tx_type)
+
         assert len(traces) == len(fee_estimation_infos) == len(external_txs)
-        return traces, fee_estimation_infos
+        return traces, fee_estimation_infos, transaction_types
 
     async def estimate_message_fee(
         self, call: CallL1Handler, block_id: BlockId = DEFAULT_BLOCK_ID

test/rpc/rpc_utils.py

@@ -11,12 +11,14 @@
     PREDEPLOYED_ACCOUNT_ADDRESS,
     PREDEPLOYED_ACCOUNT_PRIVATE_KEY,
     STORAGE_CONTRACT_PATH,
+    SUPPORTED_RPC_TX_VERSION,
 )
 from test.util import assert_transaction
 from typing import List, Union
 
 import requests
 
+from starknet_devnet.account_util import get_execute_args
 from starknet_devnet.blueprints.rpc.schema import felt_pattern_from_schema
 
 
@@ -118,3 +120,15 @@ def is_felt(value: str) -> bool:
     Check whether value is a Felt
     """
     return bool(re.match(felt_pattern_from_schema(), value))
+
+
+def get_predeployed_acc_execute_args(calls):
+    """Get execute arguments with predeployed account"""
+    return get_execute_args(
+        calls=calls,
+        account_address=PREDEPLOYED_ACCOUNT_ADDRESS,
+        private_key=PREDEPLOYED_ACCOUNT_PRIVATE_KEY,
+        nonce=0,
+        version=SUPPORTED_RPC_TX_VERSION,
+        max_fee=0,
+    )

test/rpc/test_rpc_estimate_fee.py

@@ -6,7 +6,10 @@
 
 import copy
 from test.account import _get_signature, get_nonce
-from test.rpc.rpc_utils import rpc_call_background_devnet
+from test.rpc.rpc_utils import (
+    get_predeployed_acc_execute_args,
+    rpc_call_background_devnet,
+)
 from test.rpc.test_rpc_transactions import pad_zero_entry_points
 from test.shared import (
     PREDEPLOY_ACCOUNT_CLI_ARGS,
@@ -29,7 +32,6 @@
 )
 from starkware.starknet.services.api.gateway.transaction_utils import decompress_program
 
-from starknet_devnet.account_util import get_execute_args
 from starknet_devnet.blueprints.rpc.structures.payloads import (
     RpcBroadcastedDeclareTxnV1,
     RpcBroadcastedDeclareTxnV2,
@@ -58,18 +60,6 @@ def common_estimate_response(response):
         assert overall_fee == gas_consumed * gas_price
 
 
-def get_predeployed_acc_execute_args(calls):
-    """Get execute arguments with predeployed account"""
-    return get_execute_args(
-        calls=calls,
-        account_address=PREDEPLOYED_ACCOUNT_ADDRESS,
-        private_key=PREDEPLOYED_ACCOUNT_PRIVATE_KEY,
-        nonce=0,
-        version=SUPPORTED_RPC_TX_VERSION,
-        max_fee=0,
-    )
-
-
 @pytest.mark.usefixtures("run_devnet_in_background")
 @pytest.mark.parametrize(
     "run_devnet_in_background",