fix(indexes): deps index was not initialized correctly

hathor/conf/mainnet.py

@@ -62,6 +62,8 @@
         cp(2_700_000, bytes.fromhex('00000000000000000cf3a35ab01a2281024ca4ca7871f5a6d67106eb36151038')),
         cp(2_800_000, bytes.fromhex('000000000000000004439733fd419a8a747e8afe2f89348a17c1fac24538a63c')),
         cp(2_900_000, bytes.fromhex('0000000000000000090cbd5a7958c82a2b969103001d92334f287dadcf3e01bc')),
+        cp(3_000_000, bytes.fromhex('000000000000000013c9086f4ce441f5db5de55a5e235f4f7f1ef223aedfe2db')),
+        cp(3_100_000, bytes.fromhex('00000000000000000d226a5998ffc65af89b1226126b1af1f8d0712a5301c775')),
     ],
     SOFT_VOIDED_TX_IDS=list(map(bytes.fromhex, [
         '0000000012a922a6887497bed9c41e5ed7dc7213cae107db295602168266cd02',

hathor/conf/settings.py

@@ -256,7 +256,7 @@ def MAXIMUM_NUMBER_OF_HALVINGS(self) -> int:
     PEER_MAX_CONNECTIONS: int = 125
 
     # Maximum period without receiving any messages from ther peer (in seconds).
-    PEER_IDLE_TIMEOUT: int = 60
+    PEER_IDLE_TIMEOUT: int = 600
 
     # Filepath of ca certificate file to generate connection certificates
     CA_FILEPATH: str = os.path.join(os.path.dirname(__file__), '../p2p/ca.crt')
@@ -358,9 +358,12 @@ def MAXIMUM_NUMBER_OF_HALVINGS(self) -> int:
     # List of soft voided transaction.
     SOFT_VOIDED_TX_IDS: List[bytes] = []
 
-    # Identifier used in metadata's voided_by.
+    # Identifier used in metadata's voided_by to mark a tx as soft-voided.
     SOFT_VOIDED_ID: bytes = b'tx-non-grata'
 
+    # Identifier used in metadata's voided_by to mark a tx as partially validated.
+    PARTIALLY_VALIDATED_ID: bytes = b'in-security-check'  # XXX: better name?
+
     ENABLE_EVENT_QUEUE_FEATURE: bool = False
 
     EVENT_API_DEFAULT_BATCH_SIZE: int = 100

hathor/consensus.py

@@ -103,6 +103,11 @@ def update(self, base: BaseTransaction) -> None:
         """Run a consensus update with its own context, indexes will be updated accordingly."""
         from hathor.transaction import Block, Transaction
 
+        # XXX: first make sure we can run the consensus update on this tx:
+        meta = base.get_metadata()
+        assert meta.voided_by is None or (settings.PARTIALLY_VALIDATED_ID not in meta.voided_by)
+        assert meta.validation.is_fully_connected()
+
         # this context instance will live only while this update is running
         context = self.create_context()
 

hathor/graphviz.py

@@ -52,6 +52,7 @@ def __init__(self, storage: TransactionStorage, include_funds: bool = False,
         self.voided_attrs = dict(style='dashed,filled', penwidth='0.25', fillcolor='#BDC3C7')
         self.soft_voided_attrs = dict(style='dashed,filled', penwidth='0.25', fillcolor='#CCCCFF')
         self.conflict_attrs = dict(style='dashed,filled', penwidth='2.0', fillcolor='#BDC3C7')
+        self.not_fully_validated_attrs = dict(style='dashed,filled', penwidth='0.25', fillcolor='#F9FFAB')
 
         # Labels
         self.labels: Dict[bytes, str] = {}
@@ -96,6 +97,9 @@ def get_node_attrs(self, tx: BaseTransaction) -> Dict[str, str]:
             else:
                 node_attrs.update(self.voided_attrs)
 
+        if not meta.validation.is_fully_connected():
+            node_attrs.update(self.not_fully_validated_attrs)
+
         return node_attrs
 
     def get_edge_attrs(self, tx: BaseTransaction, neighbor_hash: bytes) -> Dict[str, str]:

hathor/indexes/address_index.py

@@ -17,6 +17,7 @@
 
 from structlog import get_logger
 
+from hathor.indexes.base_index import Scope
 from hathor.indexes.tx_group_index import TxGroupIndex
 from hathor.pubsub import HathorEvents
 from hathor.transaction import BaseTransaction
@@ -26,12 +27,21 @@
 
 logger = get_logger()
 
+SCOPE = Scope(
+    include_blocks=True,
+    include_txs=True,
+    include_voided=True,
+)
+
 
 class AddressIndex(TxGroupIndex[str]):
     """ Index of inputs/outputs by address
     """
     pubsub: Optional['PubSubManager']
 
+    def get_scope(self) -> Scope:
+        return SCOPE
+
     def init_loop_step(self, tx: BaseTransaction) -> None:
         self.add_tx(tx)
 

hathor/indexes/base_index.py

@@ -13,12 +13,82 @@
 # limitations under the License.
 
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Iterator, NamedTuple, Optional
 
 from hathor.transaction.base_transaction import BaseTransaction
 
 if TYPE_CHECKING:  # pragma: no cover
     from hathor.indexes.manager import IndexesManager
+    from hathor.transaction.storage import TransactionStorage
+
+
+class Scope(NamedTuple):
+    """ This class models the scope of transactions that an index is interested in.
+
+    It is used for both selecting the optimal iterator for all the indexes that need to be initialized and for
+    filtering which transactions are fed to the index.
+    """
+    include_blocks: bool
+    include_txs: bool
+    include_voided: bool
+    # XXX: these have a default value since it should be really rare to have it different
+    include_partial: bool = False
+    topological_order: bool = True  # if False than ordering doesn't matter
+
+    # XXX: this is used to join the scope of multiple indexes to get an overall scope that includes everything that
+    #      each individual scope needs, the OR operator was chosen because it represents well the operation of keeping
+    #      a property if either A or B needs it
+    def __or__(self, other):
+        # XXX: note that this doesn't necessarily have to be OR operations between properties, we want the operations
+        #      that broaden the scope, and not narrow it.
+        # XXX: in the case of topological_order, we want to keep the "topological" ordering if any of them requires it,
+        #      so it also is an OR operator
+        return Scope(
+            include_blocks=self.include_blocks | other.include_blocks,
+            include_txs=self.include_txs | other.include_txs,
+            include_voided=self.include_voided | other.include_voided,
+            include_partial=self.include_partial | other.include_partial,
+            topological_order=self.topological_order | other.topological_order,
+        )
+
+    def matches(self, tx: BaseTransaction) -> bool:
+        """ Check if a transaction matches this scope, True means the index is interested in this transaction.
+        """
+        if tx.is_block and not self.include_blocks:
+            return False
+        if tx.is_transaction and not self.include_txs:
+            return False
+        tx_meta = tx.get_metadata()
+        if tx_meta.voided_by and not self.include_voided:
+            return False
+        if not tx_meta.validation.is_fully_connected() and not self.include_partial:
+            return False
+        # XXX: self.topologial_order doesn't affect self.match()
+        # passed all checks
+        return True
+
+    def get_iterator(self, tx_storage: 'TransactionStorage') -> Iterator[BaseTransaction]:
+        iterator: Iterator[BaseTransaction]
+        # XXX: this is to mark if the chosen iterator will yield partial transactions
+        iterator_covers_partial: bool
+        if self.topological_order:
+            iterator = tx_storage.topological_iterator()
+            iterator_covers_partial = False
+        else:
+            iterator = tx_storage.get_all_transactions()
+            iterator_covers_partial = True
+        for tx in iterator:
+            if self.matches(tx):
+                yield tx
+        if self.include_partial and not iterator_covers_partial:
+            # if partial transactions are needed and were not already covered, we use get_all_transactions, which
+            # includes partial transactions, to yield them, skipping all that aren't partial
+            for tx in tx_storage.get_all_transactions():
+                tx_meta = tx.get_metadata()
+                if tx_meta.validation.is_fully_connected():
+                    continue
+                if self.matches(tx):
+                    yield tx
 
 
 class BaseIndex(ABC):
@@ -35,6 +105,11 @@ def init_start(self, indexes_manager: 'IndexesManager') -> None:
         """
         pass
 
+    @abstractmethod
+    def get_scope(self) -> Scope:
+        """ Returns the scope of interest of this index, whether the scope is configurable is up to the index."""
+        raise NotImplementedError
+
     @abstractmethod
     def get_db_name(self) -> Optional[str]:
         """ The returned string is used to generate the relevant attributes for storing an indexe's state in the db.

hathor/indexes/deps_index.py

@@ -15,7 +15,7 @@
 from abc import abstractmethod
 from typing import TYPE_CHECKING, Iterator, List
 
-from hathor.indexes.base_index import BaseIndex
+from hathor.indexes.base_index import BaseIndex, Scope
 from hathor.transaction import BaseTransaction, Block
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -25,6 +25,13 @@
 # XXX: this arbitrary height limit must fit in a u32 (4-bytes unsigned), so it can be stored easily on rocksdb
 INF_HEIGHT: int = 2**32 - 1
 
+SCOPE = Scope(
+    include_blocks=True,
+    include_txs=True,
+    include_voided=True,
+    include_partial=True
+)
+
 
 def get_requested_from_height(tx: BaseTransaction) -> int:
     """Return the height of the block that requested (directly or indirectly) the download of this transaction.
@@ -44,7 +51,7 @@ def get_requested_from_height(tx: BaseTransaction) -> int:
         #      I'm defaulting the height to `inf` (practically), this should make it heightest priority when
         #      choosing which transactions to fetch next
         return INF_HEIGHT
-    block = tx.storage.get_transaction(first_block)
+    block = tx.storage.get_transaction(first_block, allow_partially_valid=True)
     assert isinstance(block, Block)
     return block.get_metadata().height
 
@@ -105,11 +112,15 @@ class DepsIndex(BaseIndex):
       them.
     """
 
+    def get_scope(self) -> Scope:
+        return SCOPE
+
     def init_loop_step(self, tx: BaseTransaction) -> None:
         tx_meta = tx.get_metadata()
         if tx_meta.voided_by:
             return
-        self.add_tx(tx, partial=False)
+        # self.add_tx(tx, partial=False)
+        self.add_tx(tx)
 
     def update(self, tx: BaseTransaction) -> None:
         assert tx.hash is not None

hathor/indexes/height_index.py

@@ -15,11 +15,17 @@
 from abc import abstractmethod
 from typing import List, NamedTuple, Optional, Tuple
 
-from hathor.indexes.base_index import BaseIndex
+from hathor.indexes.base_index import BaseIndex, Scope
 from hathor.transaction import BaseTransaction, Block
 from hathor.transaction.genesis import BLOCK_GENESIS
 from hathor.util import not_none
 
+SCOPE = Scope(
+    include_blocks=True,
+    include_txs=False,
+    include_voided=True,
+)
+
 
 class IndexEntry(NamedTuple):
     """Helper named tuple that implementations can use."""
@@ -40,6 +46,9 @@ class HeightIndex(BaseIndex):
     """Store the block hash for each given height
     """
 
+    def get_scope(self) -> Scope:
+        return SCOPE
+
     def init_loop_step(self, tx: BaseTransaction) -> None:
         if not tx.is_block:
             return

hathor/indexes/info_index.py

@@ -16,11 +16,19 @@
 
 from structlog import get_logger
 
-from hathor.indexes.base_index import BaseIndex
+from hathor.indexes.base_index import BaseIndex, Scope
 from hathor.transaction import BaseTransaction
 
 logger = get_logger()
 
+SCOPE = Scope(
+    include_blocks=True,
+    include_txs=True,
+    include_voided=True,
+    # XXX: this index doesn't care about the ordering
+    topological_order=False,
+)
+
 
 class InfoIndex(BaseIndex):
     """ Index of general information about the storage
@@ -30,6 +38,9 @@ def init_loop_step(self, tx: BaseTransaction) -> None:
         self.update_timestamps(tx)
         self.update_counts(tx)
 
+    def get_scope(self) -> Scope:
+        return SCOPE
+
     @abstractmethod
     def update_timestamps(self, tx: BaseTransaction) -> None:
         raise NotImplementedError

hathor/indexes/manager.py

@@ -186,7 +186,7 @@ def _manually_initialize(self, tx_storage: 'TransactionStorage') -> None:
 
         self.log.debug('indexes init')
         if indexes_to_init:
-            tx_iter = progress(tx_storage.topological_iterator(), log=self.log, total=tx_storage.get_vertices_count())
+            tx_iter = progress(tx_storage._topological_sort_dfs(), log=self.log, total=tx_storage.get_vertices_count())
         else:
             tx_iter = iter([])
         for tx in tx_iter:
@@ -307,13 +307,13 @@ def __init__(self) -> None:
         from hathor.indexes.memory_tips_index import MemoryTipsIndex
 
         self.info = MemoryInfoIndex()
-        self.all_tips = MemoryTipsIndex()
-        self.block_tips = MemoryTipsIndex()
-        self.tx_tips = MemoryTipsIndex()
+        self.all_tips = MemoryTipsIndex(all=True)
+        self.block_tips = MemoryTipsIndex(blocks=True)
+        self.tx_tips = MemoryTipsIndex(txs=True)
 
-        self.sorted_all = MemoryTimestampIndex()
-        self.sorted_blocks = MemoryTimestampIndex()
-        self.sorted_txs = MemoryTimestampIndex()
+        self.sorted_all = MemoryTimestampIndex(all=True)
+        self.sorted_blocks = MemoryTimestampIndex(blocks=True)
+        self.sorted_txs = MemoryTimestampIndex(txs=True)
 
         self.addresses = None
         self.tokens = None
@@ -340,12 +340,12 @@ def enable_utxo_index(self) -> None:
         if self.utxo is None:
             self.utxo = MemoryUtxoIndex()
 
-    def enable_deps_index(self) -> None:
+    def enable_mempool_index(self) -> None:
         from hathor.indexes.memory_mempool_tips_index import MemoryMempoolTipsIndex
         if self.mempool_tips is None:
             self.mempool_tips = MemoryMempoolTipsIndex()
 
-    def enable_mempool_index(self) -> None:
+    def enable_deps_index(self) -> None:
         from hathor.indexes.memory_deps_index import MemoryDepsIndex
         if self.deps is None:
             self.deps = MemoryDepsIndex()
@@ -362,13 +362,13 @@ def __init__(self, db: 'rocksdb.DB') -> None:
 
         self.info = RocksDBInfoIndex(self._db)
         self.height = RocksDBHeightIndex(self._db)
-        self.all_tips = PartialRocksDBTipsIndex(self._db, 'all')
-        self.block_tips = PartialRocksDBTipsIndex(self._db, 'blocks')
-        self.tx_tips = PartialRocksDBTipsIndex(self._db, 'txs')
+        self.all_tips = PartialRocksDBTipsIndex(self._db, all=True)
+        self.block_tips = PartialRocksDBTipsIndex(self._db, blocks=True)
+        self.tx_tips = PartialRocksDBTipsIndex(self._db, txs=True)
 
-        self.sorted_all = RocksDBTimestampIndex(self._db, 'all')
-        self.sorted_blocks = RocksDBTimestampIndex(self._db, 'blocks')
-        self.sorted_txs = RocksDBTimestampIndex(self._db, 'txs')
+        self.sorted_all = RocksDBTimestampIndex(self._db, all=True)
+        self.sorted_blocks = RocksDBTimestampIndex(self._db, blocks=True)
+        self.sorted_txs = RocksDBTimestampIndex(self._db, txs=True)
 
         self.addresses = None
         self.tokens = None
@@ -394,13 +394,13 @@ def enable_utxo_index(self) -> None:
         if self.utxo is None:
             self.utxo = RocksDBUtxoIndex(self._db)
 
-    def enable_deps_index(self) -> None:
+    def enable_mempool_index(self) -> None:
         from hathor.indexes.memory_mempool_tips_index import MemoryMempoolTipsIndex
         if self.mempool_tips is None:
             # XXX: use of RocksDBMempoolTipsIndex is very slow and was suspended
             self.mempool_tips = MemoryMempoolTipsIndex()
 
-    def enable_mempool_index(self) -> None:
+    def enable_deps_index(self) -> None:
         from hathor.indexes.memory_deps_index import MemoryDepsIndex
         if self.deps is None:
             # XXX: use of RocksDBDepsIndex is currently suspended until it is fixed