improvements 2

Author: jansegre

hathor/consensus/consensus.py

@@ -72,13 +72,13 @@ 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 settings.CONSENSUS_FAIL_ID in meta.voided_by:
+            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

@@ -193,3 +193,7 @@ def remove_from_needed_index(self, tx: bytes) -> None:
     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]:
+        raise NotImplementedError

hathor/indexes/memory_deps_index.py

@@ -157,6 +157,14 @@ def get_next_needed_tx(self) -> bytes:
                        height=height, needed_tx=tx.hex())
         return start_hash
 
+    def checked_next_needed_txs(self, tx_storage: 'TransactionStorage') -> 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
+
     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

hathor/manager.py

@@ -939,12 +939,11 @@ def on_new_tx(self, tx: BaseTransaction, *, conn: Optional[HathorProtocol] = Non
         """
         assert self.tx_storage.is_only_valid_allowed()
         assert tx.hash is not None
+
+        already_exists = False
         if self.tx_storage.transaction_exists(tx.hash):
             self.tx_storage.compare_bytes_with_local_tx(tx)
-            if not fails_silently:
-                raise InvalidNewTransaction('Transaction already exists {}'.format(tx.hash_hex))
-            self.log.warn('on_new_tx(): Transaction already exists', tx=tx.hash_hex)
-            return False
+            already_exists = True
 
         if tx.timestamp - self.reactor.seconds() > settings.MAX_FUTURE_TIMESTAMP_ALLOWED:
             if not fails_silently:
@@ -961,8 +960,14 @@ def on_new_tx(self, tx: BaseTransaction, *, conn: Optional[HathorProtocol] = Non
             metadata = tx.get_metadata()
         except TransactionDoesNotExist:
             if not fails_silently:
-                raise InvalidNewTransaction('missing parent')
-            self.log.warn('on_new_tx(): missing parent', tx=tx.hash_hex)
+                raise InvalidNewTransaction('cannot get metadata')
+            self.log.warn('on_new_tx(): cannot get metadata', tx=tx.hash_hex)
+            return False
+
+        if already_exists and metadata.validation.is_fully_connected():
+            if not fails_silently:
+                raise InvalidNewTransaction('Transaction already exists {}'.format(tx.hash_hex))
+            self.log.warn('on_new_tx(): Transaction already exists', tx=tx.hash_hex)
             return False
 
         if metadata.validation.is_invalid():

hathor/p2p/sync_v2/manager.py

@@ -34,14 +34,16 @@
 from hathor.transaction.exceptions import HathorError
 from hathor.transaction.storage.exceptions import TransactionDoesNotExist
 from hathor.types import VertexId
-from hathor.util import Reactor
+from hathor.util import Reactor, collect_n
 
 if TYPE_CHECKING:
     from hathor.p2p.protocol import HathorProtocol
 
 settings = HathorSettings()
 logger = get_logger()
 
+MAX_GET_TRANSACTIONS_BFS_LEN: int = 256
+
 
 class PeerState(Enum):
     ERROR = 'error'
@@ -309,7 +311,9 @@ def run_sync_transactions(self) -> None:
         assert self.tx_storage.indexes is not None
         assert self.tx_storage.indexes.deps is not None
 
-        start_hash = self.tx_storage.indexes.deps.get_next_needed_tx()
+        # 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),
+                                  MAX_GET_TRANSACTIONS_BFS_LEN)
 
         # Start with the last received block and find the best block full validated in its chain
         block = self._last_received_block
@@ -323,9 +327,9 @@ def run_sync_transactions(self) -> None:
         assert block.hash is not None
         block_height = block.get_height()
 
-        self.log.info('run sync transactions', start=start_hash.hex(), end_block_hash=block.hash.hex(),
+        self.log.info('run sync transactions', start=[i.hex() for i in needed_txs], end_block_hash=block.hash.hex(),
                       end_block_height=block_height)
-        self.send_get_transactions_bfs([start_hash], block.hash)
+        self.send_get_transactions_bfs(needed_txs, block.hash)
 
     @inlineCallbacks
     def run_sync_blocks(self) -> Generator[Any, Any, None]:
@@ -890,8 +894,14 @@ def handle_get_transactions_bfs(self, payload: str) -> None:
             self.log.warn('ignore GET-TRANSACTIONS-BFS, already streaming')
             return
         data = json.loads(payload)
+        # XXX: todo verify this limit while parsing the payload.
+        start_from = data['start_from']
+        if len(start_from) > MAX_GET_TRANSACTIONS_BFS_LEN:
+            self.log.error('too many transactions in GET-TRANSACTIONS-BFS', state=self.state)
+            self.protocol.send_error_and_close_connection('Too many transactions in GET-TRANSACTIONS-BFS')
+            return
         self.log.debug('handle_get_transactions_bfs', **data)
-        start_from = [bytes.fromhex(tx_hash_hex) for tx_hash_hex in data['start_from']]
+        start_from = [bytes.fromhex(tx_hash_hex) for tx_hash_hex in start_from]
         until_first_block = bytes.fromhex(data['until_first_block'])
         self.send_transactions_bfs(start_from, until_first_block)
 

hathor/simulator/fake_connection.py

@@ -41,7 +41,8 @@ def getPeerCertificate(self) -> X509:
 
 
 class FakeConnection:
-    def __init__(self, manager1: 'HathorManager', manager2: 'HathorManager', *, latency: float = 0):
+    def __init__(self, manager1: 'HathorManager', manager2: 'HathorManager', *, latency: float = 0,
+                 autoreconnect: bool = False):
         """
         :param: latency: Latency between nodes in seconds
         """
@@ -51,20 +52,14 @@ def __init__(self, manager1: 'HathorManager', manager2: 'HathorManager', *, late
         self.manager2 = manager2
 
         self.latency = latency
-        self.is_connected = True
-
-        self._proto1 = manager1.connections.server_factory.buildProtocol(HostnameAddress(b'fake', 0))
-        self._proto2 = manager2.connections.client_factory.buildProtocol(HostnameAddress(b'fake', 0))
-
-        self.tr1 = HathorStringTransport(self._proto2.my_peer)
-        self.tr2 = HathorStringTransport(self._proto1.my_peer)
+        self.autoreconnect = autoreconnect
+        self.is_connected = False
 
         self._do_buffering = True
         self._buf1: deque[str] = deque()
         self._buf2: deque[str] = deque()
 
-        self._proto1.makeConnection(self.tr1)
-        self._proto2.makeConnection(self.tr2)
+        self.reconnect()
 
     @property
     def proto1(self):
@@ -184,6 +179,9 @@ def run_one_step(self, debug=False, force=False):
                 if debug:
                     self.log.debug('[2->1] delivered', line=line2)
 
+        if self.autoreconnect and self._proto1.aborting and self._proto2.aborting:
+            self.reconnect()
+
         return True
 
     def run_until_empty(self, max_steps: Optional[int] = None, debug: bool = False, force: bool = False) -> None:
@@ -207,6 +205,20 @@ def disconnect(self, reason):
         self._proto2.connectionLost(reason)
         self.is_connected = False
 
+    def reconnect(self) -> None:
+        from twisted.python.failure import Failure
+        if self.is_connected:
+            self.disconnect(Failure(Exception('forced reconnection')))
+        self._buf1.clear()
+        self._buf2.clear()
+        self._proto1 = self.manager1.connections.server_factory.buildProtocol(HostnameAddress(b'fake', 0))
+        self._proto2 = self.manager2.connections.client_factory.buildProtocol(HostnameAddress(b'fake', 0))
+        self.tr1 = HathorStringTransport(self._proto2.my_peer)
+        self.tr2 = HathorStringTransport(self._proto1.my_peer)
+        self._proto1.makeConnection(self.tr1)
+        self._proto2.makeConnection(self.tr2)
+        self.is_connected = True
+
     def is_empty(self):
         if self._do_buffering and (self._buf1 or self._buf2):
             return False

tests/simulation/test_simulator.py

@@ -57,7 +57,7 @@ def test_two_nodes(self):
         gen_tx1.stop()
         gen_tx2.stop()
 
-        self.simulator.run(5 * 60)
+        self.assertTrue(self.simulator.run(600, trigger=StopWhenSynced(conn12)))
 
         self.assertTrue(conn12.is_connected)
         self.assertTipsEqual(manager1, manager2)
@@ -70,7 +70,9 @@ def test_many_miners_since_beginning(self):
         for hashpower in [10e6, 5e6, 1e6, 1e6, 1e6]:
             manager = self.create_peer()
             for node in nodes:
-                conn = FakeConnection(manager, node, latency=0.085)
+                # XXX: using autoreconnect is more realistic, but ideally it shouldn't be needed, but the test is
+                #      failing without it for some reason
+                conn = FakeConnection(manager, node, latency=0.085, autoreconnect=True)
                 self.simulator.add_connection(conn)
                 stop_triggers.append(StopWhenSynced(conn))
 
@@ -95,6 +97,7 @@ def test_new_syncing_peer(self):
         nodes = []
         miners = []
         tx_generators = []
+        stop_triggers = []
 
         manager = self.create_peer()
         nodes.append(manager)
@@ -125,8 +128,9 @@ def test_new_syncing_peer(self):
         self.log.debug('adding late node')
         late_manager = self.create_peer()
         for node in nodes:
-            conn = FakeConnection(late_manager, node, latency=0.300)
+            conn = FakeConnection(late_manager, node, latency=0.300, autoreconnect=True)
             self.simulator.add_connection(conn)
+            stop_triggers.append(StopWhenSynced(conn))
 
         self.simulator.run(600)
 
@@ -135,8 +139,9 @@ def test_new_syncing_peer(self):
         for miner in miners:
             miner.stop()
 
-        self.simulator.run(1200)
-        self.simulator.run_until_complete(1200)
+        # 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):
             self.log.debug(f'checking node {idx}')