Skip to content

Heze -- Inclusion List

Note: This document is a work-in-progress for researchers and implementers.

Introduction

These are the inclusion list specifications to implement Heze.

Containers

New containers

InclusionListEntry

1
2
3
4
@dataclass(eq=True, frozen=True)
class InclusionListEntry:
    signed_inclusion_list: SignedInclusionList
    timely: bool

InclusionListStore

1
2
3
4
@dataclass
class InclusionListStore:
    inclusion_lists: DefaultDict[Tuple[Slot, Root], Dict[ValidatorIndex, InclusionListEntry]]
    equivocators: DefaultDict[Tuple[Slot, Root], Set[ValidatorIndex]]

Helpers

New get_inclusion_list_store

1
2
3
4
5
6
7
def get_inclusion_list_store() -> InclusionListStore:
    # `cached_or_new_inclusion_list_store` is implementation and context dependent.
    # It returns the cached `InclusionListStore`; if none exists,
    # it initializes a new instance, caches it and returns it.
    inclusion_list_store = cached_or_new_inclusion_list_store()

    return inclusion_list_store

New process_inclusion_list

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
def process_inclusion_list(
    store: InclusionListStore, signed_inclusion_list: SignedInclusionList, timely: bool
) -> None:
    inclusion_list = signed_inclusion_list.message
    validator_index = inclusion_list.validator_index
    key = (inclusion_list.slot, inclusion_list.dependent_root)

    if validator_index in store.inclusion_lists[key]:
        # Mark the validator as an equivocator if it published a different inclusion list
        stored_entry = store.inclusion_lists[key][validator_index]
        stored_inclusion_list = stored_entry.signed_inclusion_list.message
        if stored_inclusion_list != inclusion_list:
            store.equivocators[key].add(validator_index)

        # Ignore an inclusion list that has already been processed
        return

    # Store the signed inclusion list and its timeliness
    store.inclusion_lists[key][validator_index] = InclusionListEntry(
        signed_inclusion_list=signed_inclusion_list,
        timely=timely,
    )

New get_inclusion_list_transactions

Note: get_inclusion_list_transactions returns a list of unique transactions from all valid and non-equivocating InclusionLists for the given slot and dependent_root. When only_timely is True, only InclusionLists received in a timely manner on the p2p network are considered; otherwise, timeliness is not considered.

Note: Inclusion lists MUST be retained for at least MIN_SLOTS_FOR_INCLUSION_LISTS_REQUESTS slots beyond their slot, after which they MAY be pruned.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
def get_inclusion_list_transactions(
    store: InclusionListStore, slot: Slot, dependent_root: Root, only_timely: bool = True
) -> Sequence[Transaction]:
    key = (slot, dependent_root)
    inclusion_lists = store.inclusion_lists[key]
    equivocators = store.equivocators[key]

    transactions: list[Transaction] = []
    for validator_index, inclusion_list in inclusion_lists.items():
        # Ignore inclusion lists from equivocators
        if validator_index in equivocators:
            continue

        # Ignore untimely inclusion lists if only timely ones are requested
        if only_timely and not inclusion_list.timely:
            continue

        transactions.extend(inclusion_list.signed_inclusion_list.message.transactions)

    # Deduplicate inclusion list transactions. Order does not need to be preserved.
    return list(set(transactions))

New get_inclusion_list_bits

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
def get_inclusion_list_bits(
    store: InclusionListStore,
    committee: InclusionListCommittee,
    slot: Slot,
    dependent_root: Root,
    only_timely: bool = True,
) -> InclusionListBits:
    key = (slot, dependent_root)
    inclusion_lists = store.inclusion_lists[key]
    equivocators = store.equivocators[key]

    validator_indices = []
    for validator_index, inclusion_list in inclusion_lists.items():
        # Ignore inclusion lists from equivocators
        if validator_index in equivocators:
            continue

        # Ignore untimely inclusion lists if only timely ones are requested
        if only_timely and not inclusion_list.timely:
            continue

        validator_indices.append(validator_index)

    return InclusionListBits(
        data=[validator_index in validator_indices for validator_index in committee]
    )

New is_inclusion_list_bits_inclusive

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
def is_inclusion_list_bits_inclusive(
    store: InclusionListStore,
    committee: InclusionListCommittee,
    slot: Slot,
    dependent_root: Root,
    inclusion_list_bits: InclusionListBits,
    only_timely: bool = True,
) -> bool:
    local_inclusion_list_bits = get_inclusion_list_bits(
        store, committee, slot, dependent_root, only_timely
    )

    for i in range(INCLUSION_LIST_COMMITTEE_SIZE):
        if local_inclusion_list_bits[i] and not inclusion_list_bits[i]:
            return False
    return True