Test Excess Data Gas Fork Transition

Documentation for tests/cancun/eip4844_blobs/test_excess_data_gas_fork_transition.py.

Generate fixtures for these test cases for Cancun with:

Cancun only:

fill -v tests/cancun/eip4844_blobs/test_excess_data_gas_fork_transition.py --fork=Cancun --evm-bin=/path/to/evm-tool-dev-version

For all forks up to and including Cancun:

fill -v tests/cancun/eip4844_blobs/test_excess_data_gas_fork_transition.py --until=Cancun --evm-bin=/path/to/evm-tool-dev-version

Tests excessDataGas and dataGasUsed block fields for EIP-4844: Shard Blob Transactions at fork transition.

Test excessDataGas and dataGasUsed block fields for EIP-4844: Shard Blob Transactions at fork transition.

Spec dataclass

Parameters from the EIP-4844 specifications as defined at https://eips.ethereum.org/EIPS/eip-4844#parameters

If the parameter is not currently used within the tests, it is commented out.

Source code in tests/cancun/eip4844_blobs/spec.py

@dataclass(frozen=True)
class Spec:
    """
    Parameters from the EIP-4844 specifications as defined at
    https://eips.ethereum.org/EIPS/eip-4844#parameters

    If the parameter is not currently used within the tests, it is commented
    out.
    """

    BLOB_TX_TYPE = 0x03
    FIELD_ELEMENTS_PER_BLOB = 4096
    BLS_MODULUS = 0x73EDA753299D7D483339D80809A1D80553BDA402FFFE5BFEFFFFFFFF00000001
    BLOB_COMMITMENT_VERSION_KZG = 1
    POINT_EVALUATION_PRECOMPILE_ADDRESS = 20
    POINT_EVALUATION_PRECOMPILE_GAS = 50_000
    MAX_DATA_GAS_PER_BLOCK = 786432
    TARGET_DATA_GAS_PER_BLOCK = 393216
    MIN_DATA_GASPRICE = 1
    DATA_GASPRICE_UPDATE_FRACTION = 3338477
    # MAX_VERSIONED_HASHES_LIST_SIZE = 2**24
    # MAX_CALLDATA_SIZE = 2**24
    # MAX_ACCESS_LIST_SIZE = 2**24
    # MAX_ACCESS_LIST_STORAGE_KEYS = 2**24
    # MAX_TX_WRAP_COMMITMENTS = 2**12
    # LIMIT_BLOBS_PER_TX = 2**12
    DATA_GAS_PER_BLOB = 2**17
    HASH_OPCODE_BYTE = 0x49
    HASH_GAS_COST = 3

    @classmethod
    def kzg_to_versioned_hash(
        cls,
        kzg_commitment: bytes | int,  # 48 bytes
        blob_commitment_version_kzg: Optional[bytes | int] = None,
    ) -> bytes:
        """
        Calculates the versioned hash for a given KZG commitment.
        """
        if blob_commitment_version_kzg is None:
            blob_commitment_version_kzg = cls.BLOB_COMMITMENT_VERSION_KZG
        if isinstance(kzg_commitment, int):
            kzg_commitment = kzg_commitment.to_bytes(48, "big")
        if isinstance(blob_commitment_version_kzg, int):
            blob_commitment_version_kzg = blob_commitment_version_kzg.to_bytes(1, "big")
        return blob_commitment_version_kzg + sha256(kzg_commitment).digest()[1:]

    @classmethod
    def fake_exponential(cls, factor: int, numerator: int, denominator: int) -> int:
        """
        Used to calculate the data gas cost.
        """
        i = 1
        output = 0
        numerator_accumulator = factor * denominator
        while numerator_accumulator > 0:
            output += numerator_accumulator
            numerator_accumulator = (numerator_accumulator * numerator) // (denominator * i)
            i += 1
        return output // denominator

    @classmethod
    def calc_excess_data_gas(cls, parent: BlockHeaderDataGasFields) -> int:
        """
        Calculate the excess data gas for a block given the excess data gas
        and data gas used from the parent block header.
        """
        if parent.excess_data_gas + parent.data_gas_used < cls.TARGET_DATA_GAS_PER_BLOCK:
            return 0
        else:
            return parent.excess_data_gas + parent.data_gas_used - cls.TARGET_DATA_GAS_PER_BLOCK

    # Note: Currently unused.
    # @classmethod
    # def get_total_data_gas(cls, tx: Transaction) -> int:
    #     """
    #     Calculate the total data gas for a transaction.
    #     """
    #     if tx.blob_versioned_hashes is None:
    #         return 0
    #     return cls.DATA_GAS_PER_BLOB * len(tx.blob_versioned_hashes)

    @classmethod
    def get_data_gasprice(cls, *, excess_data_gas: int) -> int:
        """
        Calculate the data gas price from the excess.
        """
        return cls.fake_exponential(
            cls.MIN_DATA_GASPRICE,
            excess_data_gas,
            cls.DATA_GASPRICE_UPDATE_FRACTION,
        )

SpecHelpers dataclass

Define parameters and helper functions that are tightly coupled to the 4844 spec but not strictly part of it.

Source code in tests/cancun/eip4844_blobs/spec.py

@dataclass(frozen=True)
class SpecHelpers:
    """
    Define parameters and helper functions that are tightly coupled to the 4844
    spec but not strictly part of it.
    """

    BYTES_PER_FIELD_ELEMENT = 32

    @classmethod
    def max_blobs_per_block(cls) -> int:  # MAX_BLOBS_PER_BLOCK =
        """
        Returns the maximum number of blobs per block.
        """
        return Spec.MAX_DATA_GAS_PER_BLOCK // Spec.DATA_GAS_PER_BLOB

    @classmethod
    def target_blobs_per_block(cls) -> int:
        """
        Returns the target number of blobs per block.
        """
        return Spec.TARGET_DATA_GAS_PER_BLOCK // Spec.DATA_GAS_PER_BLOB

    @classmethod
    def calc_excess_data_gas_from_blob_count(
        cls, parent_excess_data_gas: int, parent_blob_count: int
    ) -> int:
        """
        Calculate the excess data gas for a block given the parent excess data gas
        and the number of blobs in the block.
        """
        parent_consumed_data_gas = parent_blob_count * Spec.DATA_GAS_PER_BLOB
        return Spec.calc_excess_data_gas(
            BlockHeaderDataGasFields(parent_excess_data_gas, parent_consumed_data_gas)
        )

    @classmethod
    def get_min_excess_data_gas_for_data_gas_price(cls, data_gas_price: int) -> int:
        """
        Gets the minimum required excess data gas value to get a given data gas cost in a block
        """
        current_excess_data_gas = 0
        current_data_gas_price = 1
        while current_data_gas_price < data_gas_price:
            current_excess_data_gas += Spec.DATA_GAS_PER_BLOB
            current_data_gas_price = Spec.get_data_gasprice(
                excess_data_gas=current_excess_data_gas
            )
        return current_excess_data_gas

    @classmethod
    def get_min_excess_data_blobs_for_data_gas_price(cls, data_gas_price: int) -> int:
        """
        Gets the minimum required excess data blobs to get a given data gas cost in a block
        """
        return (
            cls.get_min_excess_data_gas_for_data_gas_price(data_gas_price)
            // Spec.DATA_GAS_PER_BLOB
        )

test_invalid_pre_fork_block_with_blob_fields(blockchain_test, env, pre, pre_fork_blocks, excess_data_gas_present, data_gas_used_present)

Test block rejection when excessDataGas and/or dataGasUsed fields are present on a pre-fork block.

Source code in tests/cancun/eip4844_blobs/test_excess_data_gas_fork_transition.py

@pytest.mark.parametrize(
    "excess_data_gas_present,data_gas_used_present",
    [
        (True, False),
        (False, True),
        (True, True),
    ],
)
def test_invalid_pre_fork_block_with_blob_fields(
    blockchain_test: BlockchainTestFiller,
    env: Environment,
    pre: Mapping[str, Account],
    pre_fork_blocks: List[Block],
    excess_data_gas_present: bool,
    data_gas_used_present: bool,
):
    """
    Test block rejection when `excessDataGas` and/or `dataGasUsed` fields are present on a pre-fork
    block.
    """
    header_modifier = Header()
    if excess_data_gas_present:
        header_modifier.excess_data_gas = 0
    if data_gas_used_present:
        header_modifier.data_gas_used = 0
    blockchain_test(
        pre=pre,
        post={},
        blocks=pre_fork_blocks[:-1]
        + [
            Block(
                timestamp=(FORK_TIMESTAMP - 1),
                rlp_modifier=header_modifier,
                exception="invalid field",
            )
        ],
        genesis_environment=env,
        tag="invalid_pre_fork_excess_data_gas",
    )

test_invalid_post_fork_block_without_blob_fields(blockchain_test, env, pre, pre_fork_blocks, excess_data_gas_missing, data_gas_used_missing)

Test block rejection when excessDataGas and/or dataGasUsed fields are missing on a post-fork block.

Source code in tests/cancun/eip4844_blobs/test_excess_data_gas_fork_transition.py

@pytest.mark.parametrize(
    "excess_data_gas_missing,data_gas_used_missing",
    [
        (True, False),
        (False, True),
        (True, True),
    ],
)
def test_invalid_post_fork_block_without_blob_fields(
    blockchain_test: BlockchainTestFiller,
    env: Environment,
    pre: Mapping[str, Account],
    pre_fork_blocks: List[Block],
    excess_data_gas_missing: bool,
    data_gas_used_missing: bool,
):
    """
    Test block rejection when `excessDataGas` and/or `dataGasUsed` fields are missing on a
    post-fork block.
    """
    header_modifier = Header()
    if excess_data_gas_missing:
        header_modifier.excess_data_gas = Header.REMOVE_FIELD
    if data_gas_used_missing:
        header_modifier.data_gas_used = Header.REMOVE_FIELD
    blockchain_test(
        pre=pre,
        post={},
        blocks=pre_fork_blocks
        + [
            Block(
                timestamp=FORK_TIMESTAMP,
                rlp_modifier=header_modifier,
                exception="missing field",
            )
        ],
        genesis_environment=env,
        tag="excess_data_gas_missing_post_fork",
    )

test_fork_transition_excess_data_gas(blockchain_test, env, pre, pre_fork_blocks, post_fork_blocks, post)

Test excessDataGas calculation in the header when the fork is activated.

Also produce enough blocks to test the data gas price increase when the block is full with SpecHelpers.max_blobs_per_block()bs_per_block() blobs.

Source code in tests/cancun/eip4844_blobs/test_excess_data_gas_fork_transition.py

@pytest.mark.parametrize(
    "post_fork_block_count,blob_count_per_block",
    [
        (
            SpecHelpers.get_min_excess_data_blobs_for_data_gas_price(2)
            // (SpecHelpers.max_blobs_per_block() - SpecHelpers.target_blobs_per_block())
            + 2,
            SpecHelpers.max_blobs_per_block(),
        ),
        (10, 0),
        (10, SpecHelpers.target_blobs_per_block()),
    ],
    ids=["max_blobs", "no_blobs", "target_blobs"],
)
def test_fork_transition_excess_data_gas(
    blockchain_test: BlockchainTestFiller,
    env: Environment,
    pre: Mapping[str, Account],
    pre_fork_blocks: List[Block],
    post_fork_blocks: List[Block],
    post: Mapping[str, Account],
):
    """
    Test `excessDataGas` calculation in the header when the fork is activated.

    Also produce enough blocks to test the data gas price increase when the block is full with
    `SpecHelpers.max_blobs_per_block()bs_per_block()` blobs.
    """
    blockchain_test(
        pre=pre,
        post=post,
        blocks=pre_fork_blocks + post_fork_blocks,
        genesis_environment=env,
        tag="correct_initial_data_gas_calc",
    )