refactor(verification): implement VerificationDependencies and VerificationModel [part 7/12]

hathor/verification/verification_dependencies.py

@@ -0,0 +1,115 @@
+#  Copyright 2023 Hathor Labs
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING
+
+from typing_extensions import Self
+
+from hathor.daa import DifficultyAdjustmentAlgorithm
+from hathor.reward_lock import get_spent_reward_locked_info
+from hathor.reward_lock.reward_lock import get_minimum_best_height
+from hathor.transaction import Block, Vertex
+from hathor.transaction.transaction import RewardLockedInfo, TokenInfo, Transaction
+from hathor.types import TokenUid, VertexId
+
+if TYPE_CHECKING:
+    from hathor.transaction.storage import TransactionStorage
+
+
+@dataclass(frozen=True, slots=True)
+class VertexDependencies:
+    """A dataclass of dependencies necessary for vertex verification."""
+    parents: dict[VertexId, Vertex]
+
+    @staticmethod
+    def _get_parents_from_storage(vertex: Vertex, storage: 'TransactionStorage') -> dict[VertexId, Vertex]:
+        return {vertex_id: storage.get_vertex(vertex_id) for vertex_id in vertex.parents}
+
+
+@dataclass(frozen=True, slots=True)
+class BasicBlockDependencies(VertexDependencies):
+    """A dataclass of dependencies necessary for basic block verification."""
+    daa_deps: dict[VertexId, Block] | None
+
+    @classmethod
+    def create_from_storage(
+        cls,
+        block: Block,
+        *,
+        storage: 'TransactionStorage',
+        daa: DifficultyAdjustmentAlgorithm,
+        skip_weight_verification: bool,
+    ) -> Self:
+        """Create a basic block dependencies instance using dependencies from a storage."""
+        parents = cls._get_parents_from_storage(block, storage)
+        daa_deps: dict[VertexId, Block] | None = None
+
+        if not block.is_genesis and not skip_weight_verification:
+            daa_dep_ids = daa.get_block_dependencies(block, storage.get_parent_block)
+            daa_deps = {vertex_id: storage.get_block(vertex_id) for vertex_id in daa_dep_ids}
+
+        return cls(
+            parents=parents,
+            daa_deps=daa_deps,
+        )
+
+    def get_parent_block(self) -> Block:
+        """Return the parent block of the block being verified."""
+        parent_blocks = [vertex for vertex in self.parents.values() if isinstance(vertex, Block)]
+        assert len(parent_blocks) == 1
+        return parent_blocks[0]
+
+    def get_parent_block_for_daa(self, block: Block) -> Block:
+        """A method for getting parent blocks during DAA-related verification."""
+        assert self.daa_deps is not None
+        parent_hash = block.get_block_parent_hash()
+        return self.daa_deps[parent_hash]
+
+
+@dataclass(frozen=True, slots=True)
+class BlockDependencies(VertexDependencies):
+    """A dataclass of dependencies necessary for block verification."""
+
+    @classmethod
+    def create_from_storage(cls, block: Block, *, storage: 'TransactionStorage') -> Self:
+        """Create a block dependencies instance using dependencies from a storage."""
+        parents = cls._get_parents_from_storage(block, storage)
+        return cls(parents=parents)
+
+
+@dataclass(frozen=True, slots=True)
+class TransactionDependencies(VertexDependencies):
+    """A dataclass of dependencies necessary for transaction verification."""
+    spent_txs: dict[VertexId, Vertex]
+    token_info: dict[TokenUid, TokenInfo]
+    reward_locked_info: RewardLockedInfo | None
+    best_block_height: int
+
+    @classmethod
+    def create_from_storage(cls, tx: Transaction, storage: 'TransactionStorage') -> Self:
+        """Create a transaction dependencies instance using dependencies from a storage."""
+        parents = cls._get_parents_from_storage(tx, storage)
+        spent_txs = {input_tx.tx_id: storage.get_vertex(input_tx.tx_id) for input_tx in tx.inputs}
+        token_info = tx.get_complete_token_info()
+        reward_locked_info = get_spent_reward_locked_info(tx, storage)
+        best_block_height = get_minimum_best_height(storage)
+
+        return cls(
+            parents=parents,
+            spent_txs=spent_txs,
+            token_info=token_info,
+            reward_locked_info=reward_locked_info,
+            best_block_height=best_block_height,
+        )

hathor/verification/verification_model.py

@@ -0,0 +1,165 @@
+#  Copyright 2024 Hathor Labs
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+from dataclasses import dataclass
+from typing import Generic, TypeAlias, TypeVar
+
+from typing_extensions import assert_never
+
+from hathor.daa import DifficultyAdjustmentAlgorithm
+from hathor.transaction import BaseTransaction, Block, MergeMinedBlock, TxVersion
+from hathor.transaction.storage import TransactionStorage
+from hathor.transaction.token_creation_tx import TokenCreationTransaction
+from hathor.transaction.transaction import Transaction
+from hathor.verification.verification_dependencies import (
+    BasicBlockDependencies,
+    BlockDependencies,
+    TransactionDependencies,
+)
+
+T = TypeVar('T', bound=BaseTransaction)
+BasicDepsT = TypeVar('BasicDepsT')
+DepsT = TypeVar('DepsT')
+
+
+@dataclass(frozen=True, slots=True)
+class _VerificationModel(Generic[T, BasicDepsT, DepsT]):
+    """A simple dataclass that wraps a vertex and all dependencies necessary for its verification."""
+    vertex: T
+    basic_deps: BasicDepsT
+    deps: DepsT | None
+
+
+@dataclass(frozen=True, slots=True)
+class VerificationBlock(_VerificationModel[Block, BasicBlockDependencies, BlockDependencies]):
+    """A simple dataclass that wraps a Block and all dependencies necessary for its verification."""
+
+
+@dataclass(frozen=True, slots=True)
+class VerificationMergeMinedBlock(_VerificationModel[MergeMinedBlock, BasicBlockDependencies, BlockDependencies]):
+    """A simple dataclass that wraps a MergeMinedBlock and all dependencies necessary for its verification."""
+
+
+@dataclass(frozen=True, slots=True)
+class VerificationTransaction(_VerificationModel[Transaction, None, TransactionDependencies]):
+    """A simple dataclass that wraps a Transaction and all dependencies necessary for its verification."""
+
+
+@dataclass(frozen=True, slots=True)
+class VerificationTokenCreationTransaction(
+    _VerificationModel[TokenCreationTransaction, None, TransactionDependencies]
+):
+    """A simple dataclass that wraps a TokenCreationTransaction and all dependencies necessary for its verification."""
+
+
+"""A type alias representing an union type for verification models for all vertex types."""
+VerificationModel: TypeAlias = (
+    VerificationBlock | VerificationMergeMinedBlock | VerificationTransaction | VerificationTokenCreationTransaction
+)
+
+
+def get_verification_model_from_storage(
+    vertex: BaseTransaction,
+    storage: TransactionStorage,
+    *,
+    daa: DifficultyAdjustmentAlgorithm,
+    skip_weight_verification: bool = False,
+    only_basic: bool = False
+) -> VerificationModel:
+    """Create a verification model instance for a vertex using dependencies from a storage."""
+    # We assert with type() instead of isinstance() because each subclass has a specific branch.
+    match vertex.version:
+        case TxVersion.REGULAR_BLOCK:
+            assert type(vertex) is Block
+            basic_deps, deps = _get_block_deps(
+                vertex,
+                storage=storage,
+                daa=daa,
+                skip_weight_verification=skip_weight_verification,
+                only_basic=only_basic,
+            )
+            return VerificationBlock(
+                vertex=vertex,
+                basic_deps=basic_deps,
+                deps=deps,
+            )
+
+        case TxVersion.MERGE_MINED_BLOCK:
+            assert type(vertex) is MergeMinedBlock
+            basic_deps, deps = _get_block_deps(
+                vertex,
+                storage=storage,
+                daa=daa,
+                skip_weight_verification=skip_weight_verification,
+                only_basic=only_basic,
+            )
+            return VerificationMergeMinedBlock(
+                vertex=vertex,
+                basic_deps=basic_deps,
+                deps=deps,
+            )
+
+        case TxVersion.REGULAR_TRANSACTION:
+            assert type(vertex) is Transaction
+            return VerificationTransaction(
+                vertex=vertex,
+                basic_deps=None,
+                deps=_get_tx_deps(vertex, storage=storage, only_basic=only_basic),
+            )
+
+        case TxVersion.TOKEN_CREATION_TRANSACTION:
+            assert type(vertex) is TokenCreationTransaction
+            return VerificationTokenCreationTransaction(
+                vertex=vertex,
+                basic_deps=None,
+                deps=_get_tx_deps(vertex, storage=storage, only_basic=only_basic),
+            )
+
+        case TxVersion.POA_BLOCK:
+            # TODO
+            raise NotImplementedError
+
+        case _:
+            assert_never(vertex.version)
+
+
+def _get_block_deps(
+    block: Block,
+    *,
+    storage: TransactionStorage,
+    daa: DifficultyAdjustmentAlgorithm,
+    skip_weight_verification: bool,
+    only_basic: bool
+) -> tuple[BasicBlockDependencies, BlockDependencies | None]:
+    """Create the necessary dependencies instances for a Block, using a storage."""
+    basic_deps = BasicBlockDependencies.create_from_storage(
+        block,
+        storage=storage,
+        daa=daa,
+        skip_weight_verification=skip_weight_verification,
+    )
+    deps = None
+    if not only_basic:
+        deps = BlockDependencies.create_from_storage(block, storage=storage)
+
+    return basic_deps, deps
+
+
+def _get_tx_deps(tx: Transaction, *, storage: TransactionStorage, only_basic: bool) -> TransactionDependencies | None:
+    """Create the necessary dependencies instances for a Transaction, using a storage."""
+    deps = None
+    if not only_basic:
+        deps = TransactionDependencies.create_from_storage(tx, storage)
+
+    return deps