resume streaming if max tx limit is reached

Author: msbrogli

hathor/p2p/sync_v2/agent.py

@@ -372,13 +372,17 @@ def run_sync_blocks(self) -> Generator[Any, Any, bool]:
         if partial_blocks:
             self.state = PeerState.SYNCING_TRANSACTIONS
             try:
-                yield self.start_transactions_streaming(partial_blocks)
+                reason = yield self.start_transactions_streaming(partial_blocks)
             except StreamingError as e:
                 self.log.info('tx streaming failed', reason=repr(e))
                 self.send_stop_transactions_streaming()
                 self.receiving_stream = False
                 return False
 
+            self.log.info('tx streaming finished', reason=reason)
+            while reason == StreamEnd.LIMIT_EXCEEDED:
+                reason = yield self.resume_transactions_streaming()
+
         self._blk_streaming_client = None
         self._tx_streaming_client = None
         return False
@@ -631,7 +635,10 @@ def send_next_blocks(self, start_hash: bytes, end_hash: bytes, quantity: int) ->
             self.log.debug('requested start_hash not found', start_hash=start_hash.hex())
             self.send_message(ProtocolMessages.NOT_FOUND, start_hash.hex())
             return
-        assert isinstance(blk, Block)
+        if not isinstance(blk, Block):
+            self.log.debug('request start_hash is not a block', start_hash=start_hash.hex())
+            self.send_message(ProtocolMessages.NOT_FOUND, start_hash.hex())
+            return
         assert blk.hash is not None
         # XXX: it is not an error for the other peer to request a voided block, we'll pretend it doesn't exist, butf
         blk_meta = blk.get_metadata()
@@ -801,6 +808,21 @@ def start_transactions_streaming(self, partial_blocks: list[Block]) -> Deferred[
         self.send_get_transactions_bfs(start_from, first_block_hash, last_block_hash)
         return self._tx_streaming_client.wait()
 
+    def resume_transactions_streaming(self) -> Deferred[StreamEnd]:
+        """Resume transaction streaming."""
+        idx = self._tx_streaming_client._idx
+        partial_blocks = self._tx_streaming_client.partial_blocks[idx:]
+        assert partial_blocks
+        start_from = list(self._tx_streaming_client._waiting_for)
+        first_block_hash = not_none(partial_blocks[0].hash)
+        last_block_hash = not_none(partial_blocks[-1].hash)
+        self.log.info('requesting transactions streaming',
+                      start_from=[x.hex() for x in start_from],
+                      first_block=first_block_hash.hex(),
+                      last_block=last_block_hash.hex())
+        self.send_get_transactions_bfs(start_from, first_block_hash, last_block_hash)
+        return self._tx_streaming_client.resume()
+
     def send_get_transactions_bfs(self,
                                   start_from: list[bytes],
                                   first_block_hash: bytes,

hathor/p2p/sync_v2/streamers.py

@@ -13,7 +13,7 @@
 # limitations under the License.
 
 from enum import IntFlag
-from typing import TYPE_CHECKING, Iterator, Optional, cast
+from typing import TYPE_CHECKING, Iterable, Iterator, Optional, Union, cast
 
 from structlog import get_logger
 from twisted.internet.interfaces import IConsumer, IDelayedCall, IPushProducer
@@ -39,6 +39,7 @@ class StreamEnd(IntFlag):
     LIMIT_EXCEEDED = 2
     STREAM_BECAME_VOIDED = 3  # this will happen when the current chain becomes voided while it is being sent
     TX_NOT_CONFIRMED = 4
+    INVALID_PARAMS = 5
 
     def __str__(self):
         if self is StreamEnd.END_HASH_REACHED:
@@ -191,6 +192,10 @@ def send_next(self) -> None:
 
 class TransactionsStreamingServer(_StreamingServerBase):
     """Streams all transactions confirmed by the given block, from right to left (decreasing timestamp).
+
+    If the start_from parameter is not empty, the BFS (Breadth-First Search) for the first block will commence
+    using the provided hashes. This mechanism enables streaming requests to continue from a specific point
+    should there be interruptions or issues.
     """
 
     def __init__(self,
@@ -206,28 +211,39 @@ def __init__(self,
 
         self.first_block: Block = cast(Block, self.tx_storage.get_transaction(first_block_hash))
         self.last_block: Block = cast(Block, self.tx_storage.get_transaction(last_block_hash))
+        self.start_from = start_from
 
         assert isinstance(self.first_block, Block)
         assert isinstance(self.last_block, Block)
 
-        self.current_block: Optional[Block] = self.first_block
-
         # TODO Validated start_from and use it when it's not empty.
 
+        self.current_block: Optional[Block] = self.first_block
         self.bfs = BFSOrderWalk(self.tx_storage, is_dag_verifications=True, is_dag_funds=True, is_left_to_right=False)
         self.iter = self.get_iter()
 
     def get_iter(self) -> Iterator[BaseTransaction]:
         """Return an iterator that yields all transactions confirmed by each block in sequence."""
+        root: Union[BaseTransaction, Iterable[BaseTransaction]]
+        skip_root: bool
         while self.current_block:
+            if not self.start_from:
+                root = self.current_block
+                skip_root = True
+            else:
+                root = self.start_from
+                skip_root = False
             self.log.debug('sending transactions from block',
                            block=not_none(self.current_block.hash).hex(),
-                           height=self.current_block.get_height())
-            it = self.bfs.run(self.current_block, skip_root=True)
+                           height=self.current_block.get_height(),
+                           start_from=self.start_from,
+                           skip_root=skip_root)
+            it = self.bfs.run(root, skip_root=skip_root)
             yield from it
             if self.current_block == self.last_block:
                 break
             self.current_block = self.current_block.get_next_block_best_chain()
+            self.start_from.clear()
 
     def send_next(self) -> None:
         """Push next transaction to peer."""

hathor/p2p/sync_v2/transaction_streaming_client.py

@@ -65,6 +65,12 @@ def wait(self) -> Deferred[StreamEnd]:
         """Return the deferred."""
         return self._deferred
 
+    def resume(self) -> Deferred[StreamEnd]:
+        """Resume receiving vertices."""
+        assert self._deferred.called
+        self._deferred = Deferred()
+        return self._deferred
+
     def fails(self, reason: 'StreamingError') -> None:
         """Fail the execution by resolving the deferred with an error."""
         self._deferred.errback(reason)
@@ -135,6 +141,7 @@ def handle_transactions_end(self, response_code: StreamEnd) -> None:
         """This method is called by the sync agent when a TRANSACTIONS-END message is received."""
         if self._deferred.called:
             return
+        self.log.info('transactions streaming ended', waiting_for=len(self._waiting_for))
         self._deferred.callback(response_code)
 
     def _execute_and_prepare_next(self) -> None: