feat(sync-v2): sync-v2 implemented, sync-v1 still default

Co-authored-by: Marcelo Salhab Brogliato <[email protected]>
Co-authored-by: Pedro Ferreira <[email protected]>

Author: 3 people

hathor/indexes/deps_index.py

@@ -52,7 +52,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
 

hathor/indexes/memory_deps_index.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+from functools import partial
 from typing import TYPE_CHECKING, Dict, FrozenSet, Iterator, List, Optional, Set, Tuple
 
 from structlog import get_logger
@@ -34,6 +35,9 @@ class MemoryDepsIndex(DepsIndex):
     _txs_with_deps_ready: Set[bytes]
 
     # Next to be downloaded
+    # - Key: hash of the tx to be downloaded
+    # - Value[0]: height
+    # - Value[1]: hash of the tx waiting for the download
     _needed_txs_index: Dict[bytes, Tuple[int, bytes]]
 
     def __init__(self):
@@ -49,10 +53,11 @@ def force_clear(self) -> None:
         self._needed_txs_index = {}
 
     def add_tx(self, tx: BaseTransaction, partial: bool = True) -> None:
-        assert tx.hash is not None
-        assert tx.storage is not None
         validation = tx.get_metadata().validation
         if validation.is_fully_connected():
+            # discover if new txs are ready because of this tx
+            self._update_new_deps_ready(tx)
+            # finally remove from rev deps
             self._del_from_deps_index(tx)
         elif not partial:
             raise ValueError('partial=False will only accept fully connected transactions')
@@ -63,6 +68,18 @@ def add_tx(self, tx: BaseTransaction, partial: bool = True) -> None:
     def del_tx(self, tx: BaseTransaction) -> None:
         self._del_from_deps_index(tx)
 
+    def _update_new_deps_ready(self, tx: BaseTransaction) -> None:
+        """Go over the reverse dependencies of tx and check if any of them are now ready to be validated.
+
+        This is also idempotent.
+        """
+        assert tx.hash is not None
+        assert tx.storage is not None
+        for candidate_hash in self._rev_dep_index.get(tx.hash, []):
+            candidate_tx = tx.storage.get_transaction(candidate_hash, allow_partially_valid=True)
+            if candidate_tx.is_ready_for_validation():
+                self._txs_with_deps_ready.add(candidate_hash)
+
     def _add_deps(self, tx: BaseTransaction) -> None:
         """This method is idempotent, because self.update needs it to be indempotent."""
         assert tx.hash is not None
@@ -89,12 +106,13 @@ def remove_ready_for_validation(self, tx: bytes) -> None:
         self._txs_with_deps_ready.discard(tx)
 
     def next_ready_for_validation(self, tx_storage: 'TransactionStorage', *, dry_run: bool = False) -> Iterator[bytes]:
+        get_partially_validated = partial(tx_storage.get_transaction, allow_partially_valid=True)
         if dry_run:
             cur_ready = self._txs_with_deps_ready.copy()
         else:
             cur_ready, self._txs_with_deps_ready = self._txs_with_deps_ready, set()
         while cur_ready:
-            yield from sorted(cur_ready, key=lambda tx_hash: tx_storage.get_transaction(tx_hash).timestamp)
+            yield from sorted(cur_ready, key=lambda tx_hash: get_partially_validated(tx_hash).timestamp)
             if dry_run:
                 cur_ready = self._txs_with_deps_ready - cur_ready
             else:
@@ -113,7 +131,8 @@ def _get_rev_deps(self, tx: bytes) -> FrozenSet[bytes]:
     def known_children(self, tx: BaseTransaction) -> List[bytes]:
         assert tx.hash is not None
         assert tx.storage is not None
-        it_rev_deps = map(tx.storage.get_transaction, self._get_rev_deps(tx.hash))
+        get_partially_validated = partial(tx.storage.get_transaction, allow_partially_valid=True)
+        it_rev_deps = map(get_partially_validated, self._get_rev_deps(tx.hash))
         return [not_none(rev.hash) for rev in it_rev_deps if tx.hash in rev.parents]
 
     # needed-txs-index methods:
@@ -139,6 +158,7 @@ def get_next_needed_tx(self) -> bytes:
 
     def _add_needed(self, tx: BaseTransaction) -> None:
         """This method is idempotent, because self.update needs it to be indempotent."""
+        assert tx.hash is not None
         assert tx.storage is not None
         tx_storage = tx.storage
 
@@ -153,3 +173,6 @@ def _add_needed(self, tx: BaseTransaction) -> None:
             if not tx_storage.transaction_exists(tx_hash):
                 self.log.debug('tx parent is needed', tx=tx_hash.hex())
                 self._needed_txs_index[tx_hash] = (height, not_none(tx.hash))
+
+        # also, remove the given transaction from needed, because we already have it
+        self._needed_txs_index.pop(tx.hash, None)

hathor/manager.py

@@ -454,9 +454,6 @@ def _initialize_components(self) -> None:
 
         self.log.debug('load blocks and transactions')
         for tx in self.tx_storage._topological_sort_dfs():
-            if self._full_verification:
-                tx.update_initial_metadata()
-
             assert tx.hash is not None
 
             tx_meta = tx.get_metadata()
@@ -486,15 +483,21 @@ def _initialize_components(self) -> None:
             try:
                 if self._full_verification:
                     # TODO: deal with invalid tx
+                    tx.calculate_height()
+                    tx._update_parents_children_metadata()
+
                     if tx.can_validate_full():
-                        self.tx_storage.add_to_indexes(tx)
+                        tx.calculate_min_height()
+                        if tx.is_genesis:
+                            assert tx.validate_checkpoint(self.checkpoints)
                         assert tx.validate_full(skip_block_weight_verification=skip_block_weight_verification)
+                        self.tx_storage.add_to_indexes(tx)
                         self.consensus_algorithm.update(tx)
                         self.tx_storage.indexes.update(tx)
                         if self.tx_storage.indexes.mempool_tips is not None:
                             self.tx_storage.indexes.mempool_tips.update(tx)  # XXX: move to indexes.update
                         if self.tx_storage.indexes.deps is not None:
-                            self.sync_v2_step_validations([tx])
+                            self.sync_v2_step_validations([tx], quiet=True)
                     else:
                         assert tx.validate_basic(skip_block_weight_verification=skip_block_weight_verification)
                     self.tx_storage.save_transaction(tx, only_metadata=True)
@@ -507,7 +510,9 @@ def _initialize_components(self) -> None:
                         assert self.tx_storage.indexes is not None
                         if self.tx_storage.indexes.mempool_tips:
                             self.tx_storage.indexes.mempool_tips.update(tx)
-                    self.tx_storage.add_to_indexes(tx)
+                    # XXX: refactor this in the future so the index manager decies whether to update each index
+                    if tx_meta.validation.is_fully_connected():
+                        self.tx_storage.add_to_indexes(tx)
                     if tx.is_transaction and tx_meta.voided_by:
                         self.tx_storage.del_from_indexes(tx)
             except (InvalidNewTransaction, TxValidationError):
@@ -562,11 +567,11 @@ def _initialize_components(self) -> None:
             for tx_hash in self.tx_storage.indexes.deps.iter():
                 if not self.tx_storage.transaction_exists(tx_hash):
                     continue
-                tx = self.tx_storage.get_transaction(tx_hash)
+                tx = self.tx_storage.get_transaction(tx_hash, allow_partially_valid=True)
                 if tx.get_metadata().validation.is_final():
                     depended_final_txs.append(tx)
             if self.tx_storage.indexes.deps is not None:
-                self.sync_v2_step_validations(depended_final_txs)
+                self.sync_v2_step_validations(depended_final_txs, quiet=True)
             self.log.debug('pending validations finished')
 
         best_height = self.tx_storage.get_height_best_block()
@@ -722,10 +727,10 @@ def _sync_v2_resume_validations(self) -> None:
             for tx_hash in self.tx_storage.indexes.deps.iter():
                 if not self.tx_storage.transaction_exists(tx_hash):
                     continue
-                tx = self.tx_storage.get_transaction(tx_hash)
+                tx = self.tx_storage.get_transaction(tx_hash, allow_partially_valid=True)
                 if tx.get_metadata().validation.is_final():
                     depended_final_txs.append(tx)
-            self.sync_v2_step_validations(depended_final_txs)
+            self.sync_v2_step_validations(depended_final_txs, quiet=True)
             self.log.debug('pending validations finished')
 
     def add_listen_address(self, addr: str) -> None:
@@ -1018,7 +1023,7 @@ def on_new_tx(self, tx: BaseTransaction, *, conn: Optional[HathorProtocol] = Non
         tx.storage = self.tx_storage
 
         try:
-            metadata = tx.get_metadata()
+            metadata = tx.get_metadata(allow_partial=partial)
         except TransactionDoesNotExist:
             if not fails_silently:
                 raise InvalidNewTransaction('missing parent')
@@ -1061,7 +1066,7 @@ def on_new_tx(self, tx: BaseTransaction, *, conn: Optional[HathorProtocol] = Non
                 self.tx_storage.indexes.update(tx)
                 if self.tx_storage.indexes.mempool_tips:
                     self.tx_storage.indexes.mempool_tips.update(tx)  # XXX: move to indexes.update
-                self.tx_fully_validated(tx)
+                self.tx_fully_validated(tx, quiet=quiet)
         elif sync_checkpoints:
             assert self.tx_storage.indexes.deps is not None
             metadata.children = self.tx_storage.indexes.deps.known_children(tx)
@@ -1073,7 +1078,10 @@ def on_new_tx(self, tx: BaseTransaction, *, conn: Optional[HathorProtocol] = Non
                 self.log.warn('on_new_tx(): checkpoint validation failed', tx=tx.hash_hex, exc_info=True)
                 return False
             self.tx_storage.save_transaction(tx)
+            self.tx_storage.indexes.deps.add_tx(tx)
+            self.log_new_object(tx, 'new {} partially accepted while syncing checkpoints', quiet=quiet)
         else:
+            assert self.tx_storage.indexes.deps is not None
             if isinstance(tx, Block) and not tx.has_basic_block_parent():
                 if not fails_silently:
                     raise InvalidNewTransaction('block parent needs to be at least basic-valid')
@@ -1089,38 +1097,33 @@ def on_new_tx(self, tx: BaseTransaction, *, conn: Optional[HathorProtocol] = Non
             # This needs to be called right before the save because we were adding the children
             # in the tx parents even if the tx was invalid (failing the verifications above)
             # then I would have a children that was not in the storage
-            tx.update_initial_metadata(save=False)
             self.tx_storage.save_transaction(tx)
+            self.tx_storage.indexes.deps.add_tx(tx)
+            self.log_new_object(tx, 'new {} partially accepted', quiet=quiet)
 
-        if tx.is_transaction and self.tx_storage.indexes.deps is not None:
+        if self.tx_storage.indexes.deps is not None:
             self.tx_storage.indexes.deps.remove_from_needed_index(tx.hash)
 
         if self.tx_storage.indexes.deps is not None:
             try:
-                self.sync_v2_step_validations([tx])
+                self.sync_v2_step_validations([tx], quiet=quiet)
             except (AssertionError, HathorError) as e:
                 if not fails_silently:
                     raise InvalidNewTransaction('step validations failed') from e
                 self.log.warn('on_new_tx(): step validations failed', tx=tx.hash_hex, exc_info=True)
                 return False
 
-        if not quiet:
-            ts_date = datetime.datetime.fromtimestamp(tx.timestamp)
-            now = datetime.datetime.fromtimestamp(self.reactor.seconds())
-            if tx.is_block:
-                self.log.info('new block', tx=tx, ts_date=ts_date, time_from_now=tx.get_time_from_now(now))
-            else:
-                self.log.info('new tx', tx=tx, ts_date=ts_date, time_from_now=tx.get_time_from_now(now))
-
         if propagate_to_peers:
             # Propagate to our peers.
             self.connections.send_tx_to_peers(tx)
 
         return True
 
-    def sync_v2_step_validations(self, txs: Iterable[BaseTransaction]) -> None:
+    def sync_v2_step_validations(self, txs: Iterable[BaseTransaction], *, quiet: bool) -> None:
         """ Step all validations until none can be stepped anymore.
         """
+        from functools import partial
+
         assert self.tx_storage.indexes is not None
         assert self.tx_storage.indexes.deps is not None
         # cur_txs will be empty when there are no more new txs that reached full
@@ -1129,13 +1132,14 @@ def sync_v2_step_validations(self, txs: Iterable[BaseTransaction]) -> None:
             assert ready_tx.hash is not None
             self.tx_storage.indexes.deps.remove_ready_for_validation(ready_tx.hash)
         it_next_ready = self.tx_storage.indexes.deps.next_ready_for_validation(self.tx_storage)
-        for tx in map(self.tx_storage.get_transaction, it_next_ready):
+        get_partially_validated = partial(self.tx_storage.get_transaction, allow_partially_valid=True)
+        for tx in map(get_partially_validated, it_next_ready):
             assert tx.hash is not None
             tx.update_initial_metadata()
             try:
                 # XXX: `reject_locked_reward` might not apply, partial validation is only used on sync-v2
                 # TODO: deal with `reject_locked_reward` on sync-v2
-                assert tx.validate_full(reject_locked_reward=True)
+                assert tx.validate_full(reject_locked_reward=False)
             except (AssertionError, HathorError):
                 # TODO
                 raise
@@ -1145,9 +1149,9 @@ def sync_v2_step_validations(self, txs: Iterable[BaseTransaction]) -> None:
                 self.tx_storage.indexes.update(tx)
                 if self.tx_storage.indexes.mempool_tips:
                     self.tx_storage.indexes.mempool_tips.update(tx)  # XXX: move to indexes.update
-                self.tx_fully_validated(tx)
+                self.tx_fully_validated(tx, quiet=quiet)
 
-    def tx_fully_validated(self, tx: BaseTransaction) -> None:
+    def tx_fully_validated(self, tx: BaseTransaction, *, quiet: bool) -> None:
         """ Handle operations that need to happen once the tx becomes fully validated.
 
         This might happen immediately after we receive the tx, if we have all dependencies
@@ -1166,6 +1170,30 @@ def tx_fully_validated(self, tx: BaseTransaction) -> None:
             # TODO Remove it and use pubsub instead.
             self.wallet.on_new_tx(tx)
 
+        self.log_new_object(tx, 'new {}', quiet=quiet)
+
+    def log_new_object(self, tx: BaseTransaction, message_fmt: str, *, quiet: bool) -> None:
+        """ A shortcut for logging additional information for block/txs.
+        """
+        metadata = tx.get_metadata()
+        now = datetime.datetime.fromtimestamp(self.reactor.seconds())
+        kwargs = {
+            'tx': tx,
+            'ts_date': datetime.datetime.fromtimestamp(tx.timestamp),
+            'time_from_now': tx.get_time_from_now(now),
+            'validation': metadata.validation.name,
+        }
+        if tx.is_block:
+            message = message_fmt.format('block')
+            kwargs['height'] = metadata.get_soft_height()
+        else:
+            message = message_fmt.format('tx')
+        if not quiet:
+            log_func = self.log.info
+        else:
+            log_func = self.log.debug
+        log_func(message, **kwargs)
+
     def listen(self, description: str, use_ssl: Optional[bool] = None) -> None:
         endpoint = self.connections.listen(description, use_ssl)
         # XXX: endpoint: IStreamServerEndpoint does not intrinsically have a port, but in practice all concrete cases

hathor/p2p/manager.py

@@ -93,6 +93,7 @@ def __init__(self, reactor: Reactor, my_peer: PeerId, server_factory: 'HathorSer
                  enable_sync_v1_1: bool) -> None:
         from hathor.p2p.sync_v1_1_factory import SyncV11Factory
         from hathor.p2p.sync_v1_factory import SyncV1Factory
+        from hathor.p2p.sync_v2_factory import SyncV2Factory
 
         if not (enable_sync_v1 or enable_sync_v1_1 or enable_sync_v2):
             raise TypeError(f'{type(self).__name__}() at least one sync version is required')
@@ -174,7 +175,7 @@ def __init__(self, reactor: Reactor, my_peer: PeerId, server_factory: 'HathorSer
         if enable_sync_v1_1:
             self._sync_factories[SyncVersion.V1_1] = SyncV11Factory(self)
         if enable_sync_v2:
-            self._sync_factories[SyncVersion.V2] = SyncV1Factory(self)
+            self._sync_factories[SyncVersion.V2] = SyncV2Factory(self)
 
     def disable_rate_limiter(self) -> None:
         """Disable global rate limiter."""