improvements 3

Author: jansegre

hathor/consensus/consensus.py

@@ -72,13 +72,10 @@ def create_context(self) -> ConsensusAlgorithmContext:
     def update(self, base: BaseTransaction) -> None:
         assert base.storage is not None
         assert base.storage.is_only_valid_allowed()
-        meta = base.get_metadata()
         try:
-            if meta.voided_by and settings.CONSENSUS_FAIL_ID in meta.voided_by:
-                self.log.warn('voided_by has previously failed, removing so we can revalidate', vertex=base.hash_hex)
-                meta.del_voided_by(settings.CONSENSUS_FAIL_ID)
             self._unsafe_update(base)
         except Exception:
+            meta = base.get_metadata()
             meta.add_voided_by(settings.CONSENSUS_FAIL_ID)
             assert base.storage is not None
             base.storage.save_transaction(base, only_metadata=True)

hathor/indexes/deps_index.py

@@ -190,10 +190,6 @@ def remove_from_needed_index(self, tx: bytes) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    def get_next_needed_tx(self) -> bytes:
-        """Choose the start hash for downloading the needed txs"""
-        raise NotImplementedError
-
-    # TODO: make this abstract
-    def checked_next_needed_txs(self, tx_storage: 'TransactionStorage') -> Iterator[bytes]:
+    def iter_next_needed_txs(self) -> Iterator[bytes]:
+        """Iterate over the next needed transactions."""
         raise NotImplementedError

hathor/indexes/memory_deps_index.py

@@ -147,23 +147,9 @@ def is_tx_needed(self, tx: bytes) -> bool:
     def remove_from_needed_index(self, tx: bytes) -> None:
         self._needed_txs_index.pop(tx, None)
 
-    def get_next_needed_tx(self) -> bytes:
-        # This strategy maximizes the chance to download multiple txs on the same stream
-        # find the tx with highest "height"
-        # XXX: we could cache this onto `needed_txs` so we don't have to fetch txs every time
-        # TODO: improve this by using some sorted data structure to make this better than O(n)
-        height, start_hash, tx = max((h, s, t) for t, (h, s) in self._needed_txs_index.items())
-        self.log.debug('next needed tx start', needed=len(self._needed_txs_index), start=start_hash.hex(),
-                       height=height, needed_tx=tx.hex())
-        return start_hash
-
-    def checked_next_needed_txs(self, tx_storage: 'TransactionStorage') -> Iterator[bytes]:
+    def iter_next_needed_txs(self) -> Iterator[bytes]:
         for _, (__, tx_hash) in self._needed_txs_index.items():
-            basic_tx_exists = False
-            with tx_storage.allow_partially_validated_context():
-                basic_tx_exists = tx_storage.transaction_exists(tx_hash)
-            if not basic_tx_exists:
-                yield tx_hash
+            yield tx_hash
 
     def _add_needed(self, tx: BaseTransaction) -> None:
         """This method is idempotent, because self.update needs it to be indempotent."""

hathor/indexes/rocksdb_deps_index.py

@@ -351,11 +351,6 @@ def remove_from_needed_index(self, tx: bytes) -> None:
         key_needed = self._to_key_needed(tx)
         self._db.delete((self._cf, key_needed))
 
-    def get_next_needed_tx(self) -> bytes:
-        # This strategy maximizes the chance to download multiple txs on the same stream
-        # Find the tx with highest "height"
-        # XXX: we could cache this onto `needed_txs` so we don't have to fetch txs every time
-        # TODO: improve this by using some sorted data structure to make this better than O(n)
-        height, start_hash, tx = max((h, s, t) for t, h, s in self._iter_needed())
-        self.log.debug('next needed tx start', start=start_hash.hex(), height=height, needed_tx=tx.hex())
-        return start_hash
+    def iter_next_needed_txs(self) -> Iterator[bytes]:
+        for _, __, tx_hash in self._iter_needed():
+            yield tx_hash

hathor/p2p/sync_v2/manager.py

@@ -1,4 +1,4 @@
-# Copyright 2021 Hathor Labs
+# 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.
@@ -42,7 +42,7 @@
 settings = HathorSettings()
 logger = get_logger()
 
-MAX_GET_TRANSACTIONS_BFS_LEN: int = 256
+MAX_GET_TRANSACTIONS_BFS_LEN: int = 8
 
 
 class PeerState(Enum):
@@ -312,7 +312,7 @@ def run_sync_transactions(self) -> None:
         assert self.tx_storage.indexes.deps is not None
 
         # start_hash = self.tx_storage.indexes.deps.get_next_needed_tx()
-        needed_txs, _ = collect_n(self.tx_storage.indexes.deps.checked_next_needed_txs(self.tx_storage),
+        needed_txs, _ = collect_n(self.tx_storage.indexes.deps.iter_next_needed_txs(),
                                   MAX_GET_TRANSACTIONS_BFS_LEN)
 
         # Start with the last received block and find the best block full validated in its chain

hathor/p2p/sync_v2/streamers.py

@@ -193,7 +193,6 @@ def __init__(self, node_sync: 'NodeBlockSync', start_from: list[BaseTransaction]
         self.last_block_height = 0
 
         self.bfs = BFSOrderWalk(self.storage, is_dag_verifications=True, is_dag_funds=True, is_left_to_right=False)
-        # XXX: maybe it should skip root
         self.iter = self.bfs.run(start_from, skip_root=False)
 
     def start(self) -> None:

hathor/util.py

@@ -363,6 +363,11 @@ def collect_n(it: Iterator[_T], n: int) -> tuple[list[_T], bool]:
 
     >>> collect_n(iter(range(10)), 8)
     ([0, 1, 2, 3, 4, 5, 6, 7], True)
+
+    # This also works for checking (albeit destructively, because it consumes from the itreator), if it is empty
+
+    >>> collect_n(iter(range(10)), 0)
+    ([], True)
     """
     if n < 0:
         raise ValueError(f'n must be non-negative, got {n}')

tests/p2p/test_sync_v2.py

@@ -107,7 +107,7 @@ def _run_restart_test(self, *, full_verification: bool, use_tx_storage_cache: bo
 
         manager3 = self.simulator.create_peer(builder3)
         self.assertEqual(partial_blocks, self._get_partial_blocks(manager3.tx_storage))
-        self.assertTrue(manager3.tx_storage.indexes.deps.has_needed_tx())
+        self.assertTrue(manager3.tx_storage.indexes.deps.has_needed_tx(manager3.tx_storage))
 
         conn13 = FakeConnection(manager1, manager3, latency=0.05)
         self.simulator.add_connection(conn13)

tests/simulation/test_simulator.py

@@ -139,8 +139,6 @@ def test_new_syncing_peer(self):
         for miner in miners:
             miner.stop()
 
-        # self.simulator.run(1200)
-        # self.simulator.run_until_complete(1200)
         self.assertTrue(self.simulator.run(3600, trigger=AllTriggers(stop_triggers)))
 
         for idx, node in enumerate(nodes):