changes (1)

Author: msbrogli

hathor/p2p/sync_v2/agent.py

@@ -540,6 +540,13 @@ def find_best_common_block(self,
         self.log.debug('find_best_common_block n-ary search finished', lo=lo, hi=hi)
         return lo
 
+    def on_block_complete(self, blk: Block, vertex_list: list[BaseTransaction]) -> None:
+        """This method is called when a block and its transactions are downloaded."""
+        for tx in vertex_list:
+            self.manager.on_new_tx(tx, propagate_to_peers=False, fails_silently=False)
+
+        self.manager.on_new_tx(blk, propagate_to_peers=False, fails_silently=False)
+
     def get_peer_block_hashes(self, heights: list[int]) -> Deferred[list[_HeightInfo]]:
         """ Returns the peer's block hashes in the given heights.
         """

hathor/p2p/sync_v2/blockchain_streaming_client.py

@@ -93,6 +93,15 @@ def handle_blocks(self, blk: Block) -> None:
             self.fails(TooManyVerticesReceivedError())
             return
 
+        # Run basic verification.
+        # if not blk.is_genesis:
+        #     try:
+        #         self.manager.verification_service.validate_basic(blk)
+        #     except TxValidationError as e:
+        #         self.fails(InvalidVertexError(repr(e)))
+        #         return
+
+        # Check for repeated blocks.
         assert blk.hash is not None
         is_duplicated = False
         if self.partial_vertex_exists(blk.hash):

hathor/p2p/sync_v2/mempool.py

@@ -63,6 +63,8 @@ def run(self) -> Deferred[None]:
         self._is_running = True
         self.reactor.callLater(0, self._run)
 
+        # TODO Implement a stop() and call it after N minutes.
+
         assert self._deferred is None
         self._deferred = Deferred()
         return self._deferred

hathor/p2p/sync_v2/streamers.py

@@ -215,9 +215,10 @@ def __init__(self,
         # TODO Validated start_from and use it when it's not empty.
 
         self.bfs = BFSOrderWalk(self.tx_storage, is_dag_verifications=True, is_dag_funds=True, is_left_to_right=False)
-        self.iter = self.get_next_tx()
+        self.iter = self.get_iter()
 
-    def get_next_tx(self) -> Iterator[BaseTransaction]:
+    def get_iter(self) -> Iterator[BaseTransaction]:
+        """Return an iterator that yields all transactions confirmed by each block in sequence."""
         while self.current_block:
             self.log.debug('sending transactions from block',
                            block=not_none(self.current_block.hash).hex(),
@@ -228,10 +229,6 @@ def get_next_tx(self) -> Iterator[BaseTransaction]:
                 break
             self.current_block = self.current_block.get_next_block_best_chain()
 
-    def start(self) -> None:
-        super().start()
-
-    # TODO: make this generic too?
     def send_next(self) -> None:
         """Push next transaction to peer."""
         assert self.is_running
@@ -262,7 +259,7 @@ def send_next(self) -> None:
             self.sync_agent.send_transactions_end(StreamEnd.TX_NOT_CONFIRMED)
             return
 
-        # Check if tx is confirmed by a block within range [first_block, last_block].
+        # Check if tx is confirmed by `self.current_block`.
         assert cur_metadata.first_block is not None
         assert self.current_block is not None
         if cur_metadata.first_block != self.current_block.hash:

hathor/p2p/sync_v2/transaction_streaming_client.py

@@ -20,7 +20,6 @@
 from hathor.p2p.sync_v2.exception import (
     InvalidVertexError,
     StreamingError,
-    TooManyRepeatedVerticesError,
     TooManyVerticesReceivedError,
     UnexpectedVertex,
 )
@@ -49,13 +48,9 @@ def __init__(self,
 
         self.partial_blocks = partial_blocks
 
-        # Let's keep it at "infinity" until a known issue is fixed.
-        self.max_repeated_transactions = 1_000_000
-
         self._deferred: Deferred[StreamEnd] = Deferred()
 
         self._tx_received: int = 0
-        self._tx_repeated: int = 0
 
         self._tx_max_quantity = DEFAULT_STREAMING_LIMIT
 
@@ -96,8 +91,16 @@ def handle_transaction(self, tx: BaseTransaction) -> None:
 
         self.log.debug('tx received', tx_id=tx.hash.hex())
 
-        # TODO Handle repeated transactions.
+        # Run basic verification.
+        # if not tx.is_genesis:
+        #     try:
+        #         self.manager.verification_service.validate_basic(tx)
+        #     except TxValidationError as e:
+        #         self.fails(InvalidVertexError(repr(e)))
+        #         return
 
+        # Any repeated transaction will fail this check because they will
+        # not belong to the waiting list.
         if tx.hash not in self._waiting_for:
             self.fails(UnexpectedVertex(tx.hash.hex()))
             return
@@ -118,21 +121,9 @@ def handle_transaction(self, tx: BaseTransaction) -> None:
 
         if self._tx_received % 100 == 0:
             self.log.debug('tx streaming in progress', txs_received=self._tx_received)
-        return
-
-        assert tx.hash is not None
-        is_duplicated = False
-        if self.partial_vertex_exists(tx.hash):
-            # We reached a tx we already have. Skip it.
-            self._tx_repeated += 1
-            is_duplicated = True
-            if self._tx_repeated > self.max_repeated_transactions:
-                self.log.debug('too many repeated transactions received', total_repeated=self._tx_repeated)
-                self.fails(TooManyRepeatedVerticesError())
-        if is_duplicated:
-            pass
 
     def get_missing_deps(self, tx: BaseTransaction) -> Iterator[bytes]:
+        """Return missing dependencies."""
         for dep in tx.get_all_dependencies():
             if self.tx_storage.transaction_exists(dep):
                 continue
@@ -147,18 +138,15 @@ def handle_transactions_end(self, response_code: StreamEnd) -> None:
         self._deferred.callback(response_code)
 
     def _execute_and_prepare_next(self) -> None:
-        self._buffer.sort(key=lambda vertex_id: self._db[vertex_id].timestamp)
-        for vertex_id in self._buffer:
-            tx = self._db[vertex_id]
-            try:
-                self.manager.on_new_tx(tx, propagate_to_peers=False, fails_silently=False)
-            except HathorError as e:
-                self.fails(InvalidVertexError(repr(e)))
-                return
+        """Add the block and its vertices to the DAG."""
+        assert not self._waiting_for
 
         blk = self.partial_blocks[self._idx]
+        vertex_list = [self._db[_id] for _id in self._buffer]
+        vertex_list.sort(key=lambda v: v.timestamp)
+
         try:
-            self.manager.on_new_tx(blk, propagate_to_peers=False, fails_silently=False)
+            self.sync_agent.on_block_complete(blk, vertex_list)
         except HathorError as e:
             self.fails(InvalidVertexError(repr(e)))
             return
@@ -168,15 +156,16 @@ def _execute_and_prepare_next(self) -> None:
             self._prepare_block(self.partial_blocks[self._idx])
 
     def _prepare_block(self, blk: 'Block') -> None:
-        # Reset everything for the next block
-        self._buffer = []
-        self._waiting_for = set()
-        self._db = {}
+        """Reset everything for the next block. It also adds blocks that have no dependencies."""
+        self._buffer.clear()
+        self._waiting_for.clear()
+        self._db.clear()
 
         # Add pending dependencies from block.
         for dep in blk.get_all_dependencies():
             if not self.tx_storage.transaction_exists(dep):
                 self._waiting_for.add(dep)
 
+        # If block is ready to be added then do it.
         if not self._waiting_for:
             self._execute_and_prepare_next()