Expose API to track gaps in the chain of headers

Author: cburgdorf

eth/abc.py

@@ -40,6 +40,7 @@
 )
 from eth.exceptions import VMError
 from eth.typing import (
+    BlockRange,
     BytesOrView,
     JournalDBCheckpoint,
     AccountState,
@@ -388,6 +389,14 @@ class SchemaAPI(ABC):
     """
     A class representing a database schema that maps values to lookup keys.
     """
+    @staticmethod
+    @abstractmethod
+    def make_header_chain_gaps_lookup_key() -> bytes:
+        """
+        Return the lookup key to retrieve the header chain integrity info from the database.
+        """
+        ...
+
     @staticmethod
     @abstractmethod
     def make_canonical_head_hash_lookup_key() -> bytes:
@@ -484,6 +493,18 @@ def __init__(self, db: AtomicDatabaseAPI) -> None:
         """
         ...
 
+    @abstractmethod
+    def get_header_chain_gaps(self) -> Tuple[BlockRange, ...]:
+        """
+        Return an ordered sequence of block ranges describing the integrity of the chain of
+        headers. Each block range describes a missing segment in the chain and each range is defined
+        with inclusive boundaries, meaning the first value describes the first missing block of that
+        segment and the second value describes the last missing block of the segment.
+
+        The last block range in the sequence is expected to have a block number of `-1` as the
+        right-hand value which is to say the gap is open-ended.
+        """
+
     #
     # Canonical Chain API
     #

eth/db/chain_gaps.py

@@ -0,0 +1,82 @@
+import enum
+from typing import Tuple
+
+from eth_typing import BlockNumber
+
+from eth.exceptions import GapTrackingCorrupted
+from eth.typing import BlockRange
+
+
+class GapChange(enum.Enum):
+
+    NoChange = enum.auto()
+    NewGap = enum.auto()
+    GapSplit = enum.auto()
+    GapShrink = enum.auto()
+    TailWrite = enum.auto()
+
+
+GapInfo = Tuple[GapChange, Tuple[BlockRange, ...]]
+
+
+def calculate_gaps(newly_persisted: BlockNumber, base_gaps: Tuple[BlockRange, ...]) -> GapInfo:
+
+    # If we have a fresh chain, our highest missing number can only be 1
+    highest_missing_number = 1 if base_gaps == () else base_gaps[-1][0]
+
+    if newly_persisted == highest_missing_number:
+        # This is adding a consecutive header at the very tail
+        new_last_marker = (newly_persisted + 1, -1)
+        new_gaps = base_gaps[:-1] + (new_last_marker,)
+        gap_change = GapChange.TailWrite
+    elif newly_persisted > highest_missing_number:
+        # We are creating a gap in the chain
+        gap_end = newly_persisted - 1
+        new_tail = ((highest_missing_number, gap_end), (newly_persisted + 1, -1),)
+        new_gaps = base_gaps[:-1] + new_tail
+        gap_change = GapChange.NewGap
+    elif newly_persisted < highest_missing_number:
+        # We are patching a gap which may either shrink an existing gap or divide it
+        matching_gaps = [
+            (index, pair) for index, pair in enumerate(base_gaps)
+            if newly_persisted >= pair[0] and newly_persisted <= pair[1]
+        ]
+
+        if len(matching_gaps) > 1:
+            raise GapTrackingCorrupted(
+                "Corrupted chain gap tracking",
+                f"No {newly_persisted} appears to be missing in multiple gaps",
+                f"1st gap goes from {matching_gaps[0][1][0]} to {matching_gaps[0][1][1]}"
+                f"2nd gap goes from {matching_gaps[1][1][0]} to {matching_gaps[1][1][1]}"
+            )
+        elif len(matching_gaps) == 0:
+            # Looks like we are just overwriting an existing header.
+            return GapChange.NoChange, base_gaps
+        elif len(matching_gaps) == 1:
+            gap_index, gap = matching_gaps[0]
+            if newly_persisted == gap[0] and newly_persisted == gap[1]:
+                updated_center: Tuple[Tuple[int, int], ...] = ()
+                gap_change = GapChange.GapShrink
+            elif newly_persisted == gap[0]:
+                # we are shrinking the gap at the start
+                updated_center = ((gap[0] + 1, gap[1],),)
+                gap_change = GapChange.GapShrink
+            elif newly_persisted == gap[1]:
+                # we are shrinking the gap at the tail
+                updated_center = ((gap[0], gap[1] - 1,),)
+                gap_change = GapChange.GapShrink
+            else:
+                # we are dividing the gap
+                first_new_gap = (gap[0], newly_persisted - 1)
+                second_new_gap = (newly_persisted + 1, gap[1])
+                updated_center = (first_new_gap, second_new_gap,)
+                gap_change = GapChange.GapSplit
+
+            before_gap = base_gaps[:gap_index]
+            after_gap = base_gaps[gap_index + 1:]
+            new_gaps = before_gap + updated_center + after_gap
+
+    else:
+        raise Exception("Invariant")
+
+    return gap_change, new_gaps

eth/db/header.py

@@ -29,13 +29,23 @@
 from eth.constants import (
     GENESIS_PARENT_HASH,
 )
+from eth.db.chain_gaps import (
+    calculate_gaps,
+    GapChange,
+    GapInfo,
+)
 from eth.exceptions import (
     CanonicalHeadNotFound,
     HeaderNotFound,
     ParentNotFound,
 )
 from eth.db.schema import SchemaV1
 from eth.rlp.headers import BlockHeader
+from eth.rlp.chain_gaps import (
+    decode_chain_gaps,
+    encode_chain_gaps,
+)
+from eth.typing import BlockRange
 from eth.validation import (
     validate_block_number,
     validate_word,
@@ -46,6 +56,40 @@ class HeaderDB(HeaderDatabaseAPI):
     def __init__(self, db: AtomicDatabaseAPI) -> None:
         self.db = db
 
+    def get_header_chain_gaps(self) -> Tuple[BlockRange, ...]:
+        return self._get_header_chain_gaps(self.db)
+
+    @classmethod
+    def _get_header_chain_gaps(cls, db: DatabaseAPI) -> Tuple[BlockRange, ...]:
+        try:
+            encoded_gaps = db[SchemaV1.make_header_chain_gaps_lookup_key()]
+        except KeyError:
+            return ()
+        else:
+            return decode_chain_gaps(encoded_gaps)
+
+    @classmethod
+    def _update_header_chain_gaps(
+            cls,
+            db: DatabaseAPI,
+            persisted_header: BlockHeaderAPI,
+            base_gaps: Tuple[BlockRange, ...] = None) -> GapInfo:
+
+            # If we make many updates in a row, we can avoid reloading the integrity info by
+            # continuously caching it and providing it as a parameter to this API
+            if base_gaps is None:
+                base_gaps = cls._get_header_chain_gaps(db)
+
+            gap_change, gaps = calculate_gaps(persisted_header.block_number, base_gaps)
+
+            if gap_change is not GapChange.NoChange:
+                db.set(
+                    SchemaV1.make_header_chain_gaps_lookup_key(),
+                    encode_chain_gaps(gaps)
+                )
+
+            return gap_change, gaps
+
     #
     # Canonical Chain API
     #
@@ -178,6 +222,7 @@ def _persist_checkpoint_header(
         previous_score = score - header.difficulty
         cls._set_hash_scores_to_db(db, header, previous_score)
         cls._set_as_canonical_chain_head(db, header, header.parent_hash)
+        cls._update_header_chain_gaps(db, header)
 
     @classmethod
     def _persist_header_chain(
@@ -211,6 +256,9 @@ def _persist_header_chain(
             rlp.encode(curr_chain_head),
         )
         score = cls._set_hash_scores_to_db(db, curr_chain_head, score)
+        gap_change, gaps = cls._update_header_chain_gaps(db, curr_chain_head)
+        if gap_change is GapChange.GapShrink or gap_change is GapChange.GapSplit:
+            cls._add_block_number_to_hash_lookup(db, curr_chain_head)
 
         orig_headers_seq = concat([(first_header,), headers_iterator])
         for parent, child in sliding_window(2, orig_headers_seq):
@@ -228,7 +276,9 @@ def _persist_header_chain(
             )
 
             score = cls._set_hash_scores_to_db(db, curr_chain_head, score)
-
+            gap_change, gaps = cls._update_header_chain_gaps(db, curr_chain_head, gaps)
+            if gap_change is GapChange.GapShrink or gap_change is GapChange.GapSplit:
+                cls._add_block_number_to_hash_lookup(db, curr_chain_head)
         try:
             previous_canonical_head = cls._get_canonical_head_hash(db)
             head_score = cls._get_score(db, previous_canonical_head)

eth/db/schema.py

@@ -20,6 +20,10 @@ def make_block_number_to_hash_lookup_key(block_number: BlockNumber) -> bytes:
     def make_block_hash_to_score_lookup_key(block_hash: Hash32) -> bytes:
         return b'block-hash-to-score:%s' % block_hash
 
+    @staticmethod
+    def make_header_chain_gaps_lookup_key() -> bytes:
+        return b'v1:header_chain_gaps'
+
     @staticmethod
     def make_transaction_hash_to_block_lookup_key(transaction_hash: Hash32) -> bytes:
         return b'transaction-hash-to-block:%s' % transaction_hash

eth/exceptions.py

@@ -65,6 +65,13 @@ class CanonicalHeadNotFound(PyEVMError):
     pass
 
 
+class GapTrackingCorrupted(PyEVMError):
+    """
+    Raised when the tracking of chain gaps appears to be corrupted
+    """
+    pass
+
+
 class Halt(PyEVMError):
     """
     Raised when an opcode function halts vm execution.

eth/rlp/chain_gaps.py

@@ -0,0 +1,40 @@
+from typing import Tuple, Iterable
+
+import rlp.sedes
+from eth_utils import to_tuple
+
+from eth._utils.numeric import signed_to_unsigned, unsigned_to_signed
+from eth.rlp.sedes import uint32
+from eth.typing import BlockRange
+
+chain_gaps = rlp.sedes.CountableList(rlp.sedes.List((uint32, uint32)))
+
+# Chain gaps are defined as sequence of markers that define gaps in a chain of connected
+# entities. The right hand side of the very last marker is expected to be -1, meaning the gap
+# is open-ended. E.g. (500, -1) means: Every header from number 500 upwards is missing.
+# [[first_missing, last_missing], ..., [first_missing, -1]]
+# Since RLP doesn't define signed integers, we convert the right-hand side from signed_to_unsigned
+# before entries are written and convert from unsigned_to_signed after entries are read from disk.
+
+
+@to_tuple
+def _convert_signed_to_unsigned(gaps: Tuple[BlockRange, ...]) -> Iterable[BlockRange]:
+        for pair in gaps:
+            yield (pair[0], signed_to_unsigned(pair[1]))
+
+
+@to_tuple
+def _convert_unsigned_to_signed(gaps: Tuple[BlockRange, ...]) -> Iterable[BlockRange]:
+    for pair in gaps:
+        yield (pair[0], unsigned_to_signed(pair[1]))
+
+
+def encode_chain_gaps(gaps: Tuple[BlockRange, ...]) -> bytes:
+    return rlp.encode(
+        _convert_signed_to_unsigned(gaps), sedes=chain_gaps
+    )
+
+
+def decode_chain_gaps(gaps: bytes) ->Tuple[BlockRange, ...]:
+    val = rlp.decode(gaps, sedes=chain_gaps)
+    return _convert_unsigned_to_signed(val)

eth/typing.py

@@ -41,6 +41,8 @@
 
 AccountDiff = Iterable[Tuple[Address, str, Union[int, bytes], Union[int, bytes]]]
 
+BlockRange = Tuple[int, int]
+
 GeneralState = Union[
     AccountState,
     List[Tuple[Address, Dict[str, Union[int, bytes, Dict[int, int]]]]]

newsfragments/1924.feature.rst

@@ -0,0 +1 @@
+Expose ``get_header_chain_gaps()`` API on HeaderDB to track chain gaps