State Trie

Introduction

The state trie is the structure responsible for storing .fork_types.Account objects.

Module Contents

Classes

LeafNode

Leaf node in the Merkle Trie

ExtensionNode

Extension node in the Merkle Trie

BranchNode

Branch node in the Merkle Trie

Trie

The Merkle Trie.

Functions

encode_internal_node

Encodes a Merkle Trie node into its RLP form. The RLP will then be

encode_node

Encode a Node for storage in the Merkle Trie.

copy_trie

Create a copy of trie. Since only frozen objects may be stored in tries,

trie_set

Stores an item in a Merkle Trie.

trie_get

Gets an item from the Merkle Trie.

common_prefix_length

Find the longest common prefix of two sequences.

nibble_list_to_compact

Compresses nibble-list into a standard byte array with a flag.

bytes_to_nibble_list

Converts a Bytes into to a sequence of nibbles (bytes with value < 16).

_prepare_trie

Prepares the trie for root calculation. Removes values that are empty,

root

Computes the root of a modified merkle patricia trie (MPT).

patricialize

Structural composition function.

Attributes

EMPTY_TRIE_ROOT

Node

K

V

InternalNode

Module Details

EMPTY_TRIE_ROOT

EMPTY_TRIE_ROOT
EMPTY_TRIE_ROOT = Root(
    hex_to_bytes(
        "56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"
    )
)

Node

Node
Node = Union[Account, Bytes, Transaction, Receipt, Uint, U256, None]

K

K
K = TypeVar("K", bound=Bytes)

V

V
V = TypeVar(
    "V",
    Optional[Account],
    Optional[Bytes],
    Bytes,
    Optional[Transaction],
    Optional[Receipt],
    Uint,
    U256,
)

LeafNode

Leaf node in the Merkle Trie

class LeafNode
rest_of_key :ethereum.base_types.Bytes
value :ethereum.rlp.RLP

ExtensionNode

Extension node in the Merkle Trie

class ExtensionNode
key_segment :ethereum.base_types.Bytes
subnode :ethereum.rlp.RLP

BranchNode

Branch node in the Merkle Trie

class BranchNode
subnodes :List[ethereum.rlp.RLP]
value :ethereum.rlp.RLP

InternalNode

InternalNode
InternalNode = Union[LeafNode, ExtensionNode, BranchNode]

encode_internal_node

encode_internal_node(node: Optional[InternalNode])ethereum.rlp.RLP

Encodes a Merkle Trie node into its RLP form. The RLP will then be serialized into a Bytes and hashed unless it is less that 32 bytes when serialized.

This function also accepts None, representing the absence of a node, which is encoded to b””.

Parameters

node (Optional[InternalNode]) – The node to encode.

Returns

encoded – The node encoded as RLP.

Return type

rlp.RLP

def encode_internal_node(node: Optional[InternalNode]) -> rlp.RLP:
    unencoded: rlp.RLP
    if node is None:
        unencoded = b""
    elif isinstance(node, LeafNode):
        unencoded = (
            nibble_list_to_compact(node.rest_of_key, True),
            node.value,
        )
    elif isinstance(node, ExtensionNode):
        unencoded = (
            nibble_list_to_compact(node.key_segment, False),
            node.subnode,
        )
    elif isinstance(node, BranchNode):
        unencoded = node.subnodes + [node.value]
    else:
        raise AssertionError(f"Invalid internal node type {type(node)}!")

    encoded = rlp.encode(unencoded)
    if len(encoded) < 32:
        return unencoded
    else:
        return keccak256(encoded)

encode_node

encode_node(node: Node, storage_root: Optional[ethereum.base_types.Bytes] = None)ethereum.base_types.Bytes

Encode a Node for storage in the Merkle Trie.

Currently mostly an unimplemented stub.

def encode_node(node: Node, storage_root: Optional[Bytes] = None) -> Bytes:
    if isinstance(node, Account):
        assert storage_root is not None
        return encode_account(node, storage_root)
    elif isinstance(node, (Transaction, Receipt, U256)):
        return rlp.encode(cast(rlp.RLP, node))
    elif isinstance(node, Bytes):
        return node
    else:
        raise AssertionError(
            f"encoding for {type(node)} is not currently implemented"
        )

Trie

The Merkle Trie.

class Trie

Bases: Generic[K, V]

secured :bool
default :V
_data :Dict[K, V]

copy_trie

copy_trie(trie: Trie[K, V])Trie[K, V]

Create a copy of trie. Since only frozen objects may be stored in tries, the contents are reused.

Parameters

trie (Trie) – Trie to copy.

Returns

new_trie – A copy of the trie.

Return type

Trie[K, V]

def copy_trie(trie: Trie[K, V]) -> Trie[K, V]:
    return Trie(trie.secured, trie.default, copy.copy(trie._data))

trie_set

trie_set(trie: Trie[K, V], key: K, value: V)None

Stores an item in a Merkle Trie.

This method deletes the key if value == trie.default, because the Merkle Trie represents the default value by omitting it from the trie.

Parameters
  • trie (Trie) – Trie to store in.

  • key (Bytes) – Key to lookup.

  • value (V) – Node to insert at key.

def trie_set(trie: Trie[K, V], key: K, value: V) -> None:
    if value == trie.default:
        if key in trie._data:
            del trie._data[key]
    else:
        trie._data[key] = value

trie_get

trie_get(trie: Trie[K, V], key: K)V

Gets an item from the Merkle Trie.

This method returns trie.default if the key is missing.

Parameters
  • trie – Trie to lookup in.

  • key – Key to lookup.

Returns

node – Node at key in the trie.

Return type

V

def trie_get(trie: Trie[K, V], key: K) -> V:
    return trie._data.get(key, trie.default)

common_prefix_length

common_prefix_length(a: Sequence, b: Sequence)int

Find the longest common prefix of two sequences.

def common_prefix_length(a: Sequence, b: Sequence) -> int:
    for i in range(len(a)):
        if i >= len(b) or a[i] != b[i]:
            return i
    return len(a)

nibble_list_to_compact

nibble_list_to_compact(x: ethereum.base_types.Bytes, is_leaf: bool)ethereum.base_types.Bytes

Compresses nibble-list into a standard byte array with a flag.

A nibble-list is a list of byte values no greater than 15. The flag is encoded in high nibble of the highest byte. The flag nibble can be broken down into two two-bit flags.

Highest nibble:

+---+---+----------+--------+
| _ | _ | is_leaf | parity |
+---+---+----------+--------+
  3   2      1         0

The lowest bit of the nibble encodes the parity of the length of the remaining nibbles – 0 when even and 1 when odd. The second lowest bit is used to distinguish leaf and extension nodes. The other two bits are not used.

Parameters
  • x – Array of nibbles.

  • is_leaf – True if this is part of a leaf node, or false if it is an extension node.

Returns

compressed – Compact byte array.

Return type

bytearray

def nibble_list_to_compact(x: Bytes, is_leaf: bool) -> Bytes:
    compact = bytearray()

    if len(x) % 2 == 0:  # ie even length
        compact.append(16 * (2 * is_leaf))
        for i in range(0, len(x), 2):
            compact.append(16 * x[i] + x[i + 1])
    else:
        compact.append(16 * ((2 * is_leaf) + 1) + x[0])
        for i in range(1, len(x), 2):
            compact.append(16 * x[i] + x[i + 1])

    return Bytes(compact)

bytes_to_nibble_list

bytes_to_nibble_list(bytes_: ethereum.base_types.Bytes)ethereum.base_types.Bytes

Converts a Bytes into to a sequence of nibbles (bytes with value < 16).

Parameters

bytes – The Bytes to convert.

Returns

nibble_list – The Bytes in nibble-list format.

Return type

Bytes

def bytes_to_nibble_list(bytes_: Bytes) -> Bytes:
    nibble_list = bytearray(2 * len(bytes_))
    for byte_index, byte in enumerate(bytes_):
        nibble_list[byte_index * 2] = (byte & 0xF0) >> 4
        nibble_list[byte_index * 2 + 1] = byte & 0x0F
    return Bytes(nibble_list)

_prepare_trie

_prepare_trie(trie: Trie[K, V], get_storage_root: Callable[[ethereum.frontier.fork_types.Address], ethereum.frontier.fork_types.Root] = None)Mapping[ethereum.base_types.Bytes, ethereum.base_types.Bytes]

Prepares the trie for root calculation. Removes values that are empty, hashes the keys (if secured == True) and encodes all the nodes.

Parameters
  • trie – The Trie to prepare.

  • get_storage_root – Function to get the storage root of an account. Needed to encode Account objects.

Returns

out – Object with keys mapped to nibble-byte form.

Return type

Mapping[ethereum.base_types.Bytes, Node]

def _prepare_trie(
    trie: Trie[K, V],
    get_storage_root: Callable[[Address], Root] = None,
) -> Mapping[Bytes, Bytes]:
    mapped: MutableMapping[Bytes, Bytes] = {}

    for (preimage, value) in trie._data.items():
        if isinstance(value, Account):
            assert get_storage_root is not None
            address = Address(preimage)
            encoded_value = encode_node(value, get_storage_root(address))
        else:
            encoded_value = encode_node(value)
        # Empty values are represented by their absence
        ensure(encoded_value != b"", AssertionError)
        key: Bytes
        if trie.secured:
            # "secure" tries hash keys once before construction
            key = keccak256(preimage)
        else:
            key = preimage
        mapped[bytes_to_nibble_list(key)] = encoded_value

    return mapped

root

root(trie: Trie[K, V], get_storage_root: Callable[[ethereum.frontier.fork_types.Address], ethereum.frontier.fork_types.Root] = None)ethereum.frontier.fork_types.Root

Computes the root of a modified merkle patricia trie (MPT).

Parameters
  • trieTrie to get the root of.

  • get_storage_root – Function to get the storage root of an account. Needed to encode Account objects.

Returns

root – MPT root of the underlying key-value pairs.

Return type

.fork_types.Root

def root(
    trie: Trie[K, V],
    get_storage_root: Callable[[Address], Root] = None,
) -> Root:
    obj = _prepare_trie(trie, get_storage_root)

    root_node = encode_internal_node(patricialize(obj, Uint(0)))
    if len(rlp.encode(root_node)) < 32:
        return keccak256(rlp.encode(root_node))
    else:
        assert isinstance(root_node, Bytes)
        return Root(root_node)

patricialize

patricialize(obj: Mapping[ethereum.base_types.Bytes, ethereum.base_types.Bytes], level: ethereum.base_types.Uint)Optional[InternalNode]

Structural composition function.

Used to recursively patricialize and merkleize a dictionary. Includes memoization of the tree structure and hashes.

Parameters
  • obj – Underlying trie key-value pairs, with keys in nibble-list format.

  • level – Current trie level.

Returns

node – Root node of obj.

Return type

ethereum.base_types.Bytes

def patricialize(
    obj: Mapping[Bytes, Bytes], level: Uint
) -> Optional[InternalNode]:
    if len(obj) == 0:
        return None

    arbitrary_key = next(iter(obj))

    # if leaf node
    if len(obj) == 1:
        leaf = LeafNode(arbitrary_key[level:], obj[arbitrary_key])
        return leaf

    # prepare for extension node check by finding max j such that all keys in
    # obj have the same key[i:j]
    substring = arbitrary_key[level:]
    prefix_length = len(substring)
    for key in obj:
        prefix_length = min(
            prefix_length, common_prefix_length(substring, key[level:])
        )

        # finished searching, found another key at the current level
        if prefix_length == 0:
            break

    # if extension node
    if prefix_length > 0:
        prefix = arbitrary_key[level : level + prefix_length]
        return ExtensionNode(
            prefix,
            encode_internal_node(patricialize(obj, level + prefix_length)),
        )

    branches: List[MutableMapping[Bytes, Bytes]] = []
    for _ in range(16):
        branches.append({})
    value = b""
    for key in obj:
        if len(key) == level:
            # shouldn't ever have an account or receipt in an internal node
            if isinstance(obj[key], (Account, Receipt, Uint)):
                raise AssertionError
            value = obj[key]
        else:
            branches[key[level]][key] = obj[key]

    return BranchNode(
        [
            encode_internal_node(patricialize(branches[k], level + 1))
            for k in range(16)
        ],
        value,
    )