review changes

Author: msbrogli

hathor/p2p/sync_v2/agent.py

@@ -211,7 +211,8 @@ def stop(self) -> None:
         if not self._started:
             raise Exception('NodeSyncBlock is already stopped')
         self._started = False
-        self._lc_run.stop()
+        if self._lc_run.running:
+            self._lc_run.stop()
 
     def get_cmd_dict(self) -> dict[ProtocolMessages, Callable[[str], None]]:
         """ Return a dict of messages of the plugin.
@@ -277,6 +278,9 @@ def run_sync(self) -> Generator[Any, Any, None]:
         self._is_running = True
         try:
             yield self._run_sync()
+        except Exception:
+            self.protocol.send_error_and_close_connection('internal error')
+            self.log.error('unhandled exception', exc_info=True)
         finally:
             self._is_running = False
 
@@ -796,7 +800,9 @@ def handle_best_block(self, payload: str) -> None:
 
     def start_transactions_streaming(self, partial_blocks: list[Block]) -> Deferred[StreamEnd]:
         """Request peer to start streaming transactions to us."""
-        self._tx_streaming_client = TransactionStreamingClient(self, partial_blocks)
+        self._tx_streaming_client = TransactionStreamingClient(self,
+                                                               partial_blocks,
+                                                               limit=self.DEFAULT_STREAMING_LIMIT)
 
         start_from: list[bytes] = []
         first_block_hash = not_none(partial_blocks[0].hash)

hathor/p2p/sync_v2/transaction_streaming_client.py

@@ -17,13 +17,8 @@
 from structlog import get_logger
 from twisted.internet.defer import Deferred
 
-from hathor.p2p.sync_v2.exception import (
-    InvalidVertexError,
-    StreamingError,
-    TooManyVerticesReceivedError,
-    UnexpectedVertex,
-)
-from hathor.p2p.sync_v2.streamers import DEFAULT_STREAMING_LIMIT, StreamEnd
+from hathor.p2p.sync_v2.exception import InvalidVertexError, StreamingError, TooManyVerticesReceivedError
+from hathor.p2p.sync_v2.streamers import StreamEnd
 from hathor.transaction import BaseTransaction
 from hathor.transaction.exceptions import HathorError, TxValidationError
 from hathor.types import VertexId
@@ -38,7 +33,9 @@
 class TransactionStreamingClient:
     def __init__(self,
                  sync_agent: 'NodeBlockSync',
-                 partial_blocks: list['Block']) -> None:
+                 partial_blocks: list['Block'],
+                 *,
+                 limit: int) -> None:
         self.sync_agent = sync_agent
         self.protocol = self.sync_agent.protocol
         self.tx_storage = self.sync_agent.tx_storage
@@ -52,7 +49,7 @@ def __init__(self,
 
         self._tx_received: int = 0
 
-        self._tx_max_quantity = DEFAULT_STREAMING_LIMIT
+        self._tx_max_quantity = limit
 
         self._idx: int = 0
         self._buffer: list[VertexId] = []
@@ -68,6 +65,7 @@ def wait(self) -> Deferred[StreamEnd]:
     def resume(self) -> Deferred[StreamEnd]:
         """Resume receiving vertices."""
         assert self._deferred.called
+        self._tx_received = 0
         self._deferred = Deferred()
         return self._deferred
 
@@ -95,15 +93,21 @@ def handle_transaction(self, tx: BaseTransaction) -> None:
         # Run basic verification.
         if not tx.is_genesis:
             try:
-                self.manager.verification_service.validate_basic(tx)
+                self.manager.verification_service.verify_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()))
+            if tx.hash in self._db:
+                # This case might happen during a resume, so we just log and keep syncing.
+                self.log.info('duplicated vertex received', tx_id=tx.hash.hex())
+            else:
+                # TODO Uncomment the following code to fail on receiving unexpected vertices.
+                # self.fails(UnexpectedVertex(tx.hash.hex()))
+                self.log.info('unexpected vertex received', tx_id=tx.hash.hex())
             return
         self._waiting_for.remove(tx.hash)
 

tests/p2p/test_sync_v2.py

@@ -219,17 +219,19 @@ def test_exceeds_streaming_and_mempool_limits(self) -> None:
         # Let the connection start to sync.
         self.simulator.run(1)
 
+        new_streaming_limit = 30
+
         # Change manager1 default streaming and mempool limits.
         sync1 = conn12.proto1.state.sync_agent
-        sync1.DEFAULT_STREAMING_LIMIT = 30
-        sync1.mempool_manager.MAX_STACK_LENGTH = 30
+        sync1.DEFAULT_STREAMING_LIMIT = new_streaming_limit
+        sync1.mempool_manager.MAX_STACK_LENGTH = new_streaming_limit
         self.assertIsNone(sync1._blk_streaming_server)
         self.assertIsNone(sync1._tx_streaming_server)
 
         # Change manager2 default streaming and mempool limits.
         sync2 = conn12.proto2.state.sync_agent
-        sync2.DEFAULT_STREAMING_LIMIT = 50
-        sync2.mempool_manager.MAX_STACK_LENGTH = 50
+        sync2.DEFAULT_STREAMING_LIMIT = new_streaming_limit
+        sync2.mempool_manager.MAX_STACK_LENGTH = new_streaming_limit
         self.assertIsNone(sync2._blk_streaming_server)
         self.assertIsNone(sync2._tx_streaming_server)