test_bls12_381_uncachable()¶

@pytest.mark.repricing
@pytest.mark.parametrize(
    "precompile_address,ret_size,generate_calldata,target",
    [
        pytest.param(
            bls12381_spec.Spec.G1ADD,
            128,
            _g1add_calldata,
            Precompile.BLS12_G1ADD,
            id="bls12_g1add",
        ),
        pytest.param(
            bls12381_spec.Spec.G2ADD,
            256,
            _g2add_calldata,
            Precompile.BLS12_G2ADD,
            id="bls12_g2add",
        ),
        pytest.param(
            bls12381_spec.Spec.G1MSM,
            128,
            _g1msm_calldata,
            Precompile.BLS12_G1MSM,
            id="bls12_g1msm",
        ),
        pytest.param(
            bls12381_spec.Spec.G2MSM,
            256,
            _g2msm_calldata,
            Precompile.BLS12_G2MSM,
            id="bls12_g2msm",
        ),
        pytest.param(
            bls12381_spec.Spec.MAP_FP_TO_G1,
            64,
            _fp_to_g1_calldata,
            Precompile.BLS12_MAP_FP_TO_G1,
            id="bls12_fp_to_g1",
        ),
        pytest.param(
            bls12381_spec.Spec.MAP_FP2_TO_G2,
            128,
            _fp2_to_g2_calldata,
            Precompile.BLS12_MAP_FP2_TO_G2,
            id="bls12_fp_to_g2",
        ),
    ],
)
def test_bls12_381_uncachable(
    benchmark_test: BenchmarkTestFiller,
    pre: Alloc,
    fork: Fork,
    gas_benchmark_value: int,
    tx_gas_limit: int,
    precompile_address: Address,
    ret_size: int,
    generate_calldata: Callable[[int], Bytes],
    target: OpcodeTarget,
) -> None:
    """
    Benchmark BLS12_381 with unique input per call.

    Each iteration writes the precompile output back to the
    start of the input so every call receives a distinct input,
    avoiding precompile result caching in clients.
    """
    if precompile_address not in fork.precompiles():
        pytest.skip("Precompile not enabled")

    intrinsic_gas_calculator = fork.transaction_intrinsic_cost_calculator()
    gas_calc_map = build_gas_calculation_function_map(fork.gas_costs())
    input_size = len(generate_calldata(0))
    precompile_cost = gas_calc_map[int(precompile_address)](input_size)

    attack_block = Op.POP(
        Op.STATICCALL(
            gas=Op.GAS,
            address=precompile_address,
            args_size=Op.CALLDATASIZE,
            ret_size=ret_size,
            # gas accounting
            address_warm=True,
            inner_call_cost=precompile_cost,
        ),
    )

    setup = Op.CALLDATACOPY(
        0,
        0,
        Op.CALLDATASIZE,
        # gas accounting
        data_size=input_size,
        new_memory_size=input_size,
    )
    setup_cost = setup.gas_cost(fork)

    loop = WhileGas(body=attack_block, fork=fork)

    code = setup + loop
    attack_contract_address = pre.deploy_contract(code=code)

    iteration_cost = loop.gas_cost(fork)

    txs: list[Transaction] = []
    remaining_gas = gas_benchmark_value
    seed = 0
    expected_opcode_count = 0
    while remaining_gas > 0:
        per_tx_gas = min(tx_gas_limit, remaining_gas)
        calldata = generate_calldata(seed)

        intrinsic = intrinsic_gas_calculator(
            calldata=calldata,
            return_cost_deducted_prior_execution=True,
        )
        gas_for_loop = per_tx_gas - intrinsic - setup_cost
        if gas_for_loop < iteration_cost:
            break
        expected_opcode_count += gas_for_loop // iteration_cost

        txs.append(
            Transaction(
                to=attack_contract_address,
                sender=pre.fund_eoa(),
                gas_limit=per_tx_gas,
                data=calldata,
            )
        )
        remaining_gas -= per_tx_gas
        seed += 1

    assert len(txs) != 0, "No transactions were added to the test."

    benchmark_test(
        target_opcode=target,
        skip_gas_used_validation=True,
        verifications=[ReceiptStatusExpected()],
        blocks=[Block(txs=txs)],
        expected_opcode_count=expected_opcode_count,
    )