refactor(consensus): Split on_new_vertex() and remove fails_silently

hathor/cli/db_import.py

@@ -90,7 +90,7 @@ def _import_txs(self) -> Iterator['BaseTransaction']:
             tx = parser.deserialize(tx_bytes)
             assert tx is not None
             tx.storage = self.tx_storage
-            self.manager.on_new_tx(tx, quiet=True, fails_silently=False)
+            self.manager.on_new_tx(tx, quiet=True)
             yield tx
 
 

hathor/cli/events_simulator/scenario.py

@@ -71,13 +71,13 @@ def simulate_single_chain_blocks_and_transactions(simulator: 'Simulator', manage
     tx = gen_new_tx(manager, address, 1000)
     tx.weight = manager.daa.minimum_tx_weight(tx)
     tx.update_hash()
-    assert manager.propagate_tx(tx, fails_silently=False)
+    assert manager.propagate_tx(tx)
     simulator.run(60)
 
     tx = gen_new_tx(manager, address, 2000)
     tx.weight = manager.daa.minimum_tx_weight(tx)
     tx.update_hash()
-    assert manager.propagate_tx(tx, fails_silently=False)
+    assert manager.propagate_tx(tx)
     simulator.run(60)
 
     add_new_blocks(manager, 1)
@@ -117,15 +117,15 @@ def simulate_unvoided_transaction(simulator: 'Simulator', manager: 'HathorManage
     tx = gen_new_tx(manager, address, 1000)
     tx.weight = 19.0005
     tx.update_hash()
-    assert manager.propagate_tx(tx, fails_silently=False)
+    assert manager.propagate_tx(tx)
     simulator.run(60)
 
     # A clone is created with a greater timestamp and a lower weight. It's a voided twin tx.
     tx2 = tx.clone(include_metadata=False)
     tx2.timestamp += 60
     tx2.weight = 19
     tx2.update_hash()
-    assert manager.propagate_tx(tx2, fails_silently=False)
+    assert manager.propagate_tx(tx2)
     simulator.run(60)
 
     # Only the second tx is voided
@@ -140,7 +140,7 @@ def simulate_unvoided_transaction(simulator: 'Simulator', manager: 'HathorManage
         tx2.hash,
     ]
     block.update_hash()
-    assert manager.propagate_tx(block, fails_silently=False)
+    assert manager.propagate_tx(block)
     simulator.run(60)
 
     # The first tx gets voided and the second gets unvoided
@@ -165,7 +165,7 @@ def simulate_invalid_mempool_transaction(simulator: 'Simulator', manager: 'Hatho
     tx = gen_new_tx(manager, address, 1000)
     tx.weight = manager.daa.minimum_tx_weight(tx)
     tx.update_hash()
-    assert manager.propagate_tx(tx, fails_silently=False)
+    assert manager.propagate_tx(tx)
     simulator.run(60)
     balance_per_address = manager.wallet.get_balance_per_address(settings.HATHOR_TOKEN_UID)
     assert balance_per_address[address] == 1000
@@ -176,7 +176,7 @@ def simulate_invalid_mempool_transaction(simulator: 'Simulator', manager: 'Hatho
     b0: Block = tb0.generate_mining_block(manager.rng, storage=manager.tx_storage)
     b0.weight = 10
     manager.cpu_mining_service.resolve(b0)
-    assert manager.propagate_tx(b0, fails_silently=False)
+    assert manager.propagate_tx(b0)
     simulator.run(60)
 
     # the transaction should have been removed from the mempool and the storage after the re-org
@@ -204,15 +204,15 @@ def simulate_empty_script(simulator: 'Simulator', manager: 'HathorManager') -> N
     tx1.outputs[1].script = b''
     tx1.weight = manager.daa.minimum_tx_weight(tx1)
     tx1.update_hash()
-    assert manager.propagate_tx(tx1, fails_silently=False)
+    assert manager.propagate_tx(tx1)
     simulator.run(60)
 
     tx2 = gen_new_tx(manager, address, 1000)
     tx2.inputs = [TxInput(tx_id=tx1.hash, index=1, data=b'\x51')]
     tx2.outputs = [TxOutput(value=1000, script=original_script)]
     tx2.weight = manager.daa.minimum_tx_weight(tx2)
     tx2.update_hash()
-    assert manager.propagate_tx(tx2, fails_silently=False)
+    assert manager.propagate_tx(tx2)
     simulator.run(60)
 
     add_new_blocks(manager, 1)
@@ -242,15 +242,15 @@ def simulate_custom_script(simulator: 'Simulator', manager: 'HathorManager') ->
     tx1.outputs[1].script = s.data
     tx1.weight = manager.daa.minimum_tx_weight(tx1)
     tx1.update_hash()
-    assert manager.propagate_tx(tx1, fails_silently=False)
+    assert manager.propagate_tx(tx1)
     simulator.run(60)
 
     tx2 = gen_new_tx(manager, address, 1000)
     tx2.inputs = [TxInput(tx_id=tx1.hash, index=1, data=bytes([len(some_data)]) + some_data)]
     tx2.outputs = [TxOutput(value=1000, script=original_script)]
     tx2.weight = manager.daa.minimum_tx_weight(tx2)
     tx2.update_hash()
-    assert manager.propagate_tx(tx2, fails_silently=False)
+    assert manager.propagate_tx(tx2)
     simulator.run(60)
 
     add_new_blocks(manager, 1)

hathor/consensus/poa/poa_block_producer.py

@@ -193,7 +193,7 @@ def _produce_block(self, previous_block: PoaBlock) -> None:
             parent=block.get_block_parent_hash().hex(),
             voided=bool(block.get_metadata().voided_by),
         )
-        self.manager.on_new_tx(block, propagate_to_peers=True, fails_silently=False)
+        self.manager.on_new_tx(block, propagate_to_peers=True)
 
     def _expected_block_timestamp(self, previous_block: Block, signer_index: int) -> int:
         """Calculate the expected timestamp for a new block."""

hathor/dag_builder/artifacts.py

@@ -63,7 +63,7 @@ def propagate_with(self, manager: HathorManager, *, up_to: str | None = None) ->
 
         for node, vertex in self.list:
             if found_begin:
-                assert manager.on_new_tx(vertex, fails_silently=False)
+                assert manager.on_new_tx(vertex)
                 self._last_propagated = node.name
 
             if node.name == self._last_propagated:

hathor/manager.py

@@ -707,7 +707,7 @@ def get_tokens_issued_per_block(self, height: int) -> int:
         """Return the number of tokens issued (aka reward) per block of a given height."""
         return self.daa.get_tokens_issued_per_block(height)
 
-    def submit_block(self, blk: Block, fails_silently: bool = True) -> bool:
+    def submit_block(self, blk: Block) -> bool:
         """Used by submit block from all mining APIs.
         """
         tips = self.tx_storage.get_best_block_tips()
@@ -724,7 +724,7 @@ def submit_block(self, blk: Block, fails_silently: bool = True) -> bool:
         )
         if blk.weight <= min_insignificant_weight:
             self.log.warn('submit_block(): insignificant weight? accepted anyway', blk=blk.hash_hex, weight=blk.weight)
-        return self.propagate_tx(blk, fails_silently=fails_silently)
+        return self.propagate_tx(blk)
 
     def push_tx(self, tx: Transaction, allow_non_standard_script: bool = False,
                 max_output_script_size: int | None = None) -> None:
@@ -755,9 +755,9 @@ def push_tx(self, tx: Transaction, allow_non_standard_script: bool = False,
         if not tx_from_lib.is_standard(max_output_script_size, not allow_non_standard_script):
             raise NonStandardTxError('Transaction is non standard.')
 
-        self.propagate_tx(tx, fails_silently=False)
+        self.propagate_tx(tx)
 
-    def propagate_tx(self, tx: BaseTransaction, fails_silently: bool = True) -> bool:
+    def propagate_tx(self, tx: BaseTransaction) -> bool:
         """Push a new transaction to the network. It is used by both the wallet and the mining modules.
 
         :return: True if the transaction was accepted
@@ -768,33 +768,26 @@ def propagate_tx(self, tx: BaseTransaction, fails_silently: bool = True) -> bool
         else:
             tx.storage = self.tx_storage
 
-        return self.on_new_tx(tx, fails_silently=fails_silently, propagate_to_peers=True)
+        return self.on_new_tx(tx, propagate_to_peers=True)
 
     def on_new_tx(
         self,
-        tx: BaseTransaction,
+        vertex: BaseTransaction,
         *,
         quiet: bool = False,
-        fails_silently: bool = True,
         propagate_to_peers: bool = True,
         reject_locked_reward: bool = True
     ) -> bool:
         """ New method for adding transactions or blocks that steps the validation state machine.
 
         :param tx: transaction to be added
         :param quiet: if True will not log when a new tx is accepted
-        :param fails_silently: if False will raise an exception when tx cannot be added
         :param propagate_to_peers: if True will relay the tx to other peers if it is accepted
         """
-        success = self.vertex_handler.on_new_vertex(
-            tx,
-            quiet=quiet,
-            fails_silently=fails_silently,
-            reject_locked_reward=reject_locked_reward,
-        )
+        success = self.vertex_handler.on_new_relayed_vertex(vertex, reject_locked_reward=reject_locked_reward)
 
         if propagate_to_peers and success:
-            self.connections.send_tx_to_peers(tx)
+            self.connections.send_tx_to_peers(vertex)
 
         return success
 

hathor/p2p/sync_v2/agent.py

@@ -22,7 +22,7 @@
 
 from structlog import get_logger
 from twisted.internet.defer import Deferred, inlineCallbacks
-from twisted.internet.task import LoopingCall, deferLater
+from twisted.internet.task import LoopingCall
 
 from hathor.conf.settings import HathorSettings
 from hathor.exception import InvalidNewTransaction
@@ -613,17 +613,11 @@ def find_best_common_block(self,
         return lo
 
     @inlineCallbacks
-    def on_block_complete(self, blk: Block, vertex_list: list[BaseTransaction]) -> Generator[Any, Any, None]:
+    def on_block_complete(self, blk: Block, vertex_list: list[Transaction]) -> Generator[Any, Any, None]:
         """This method is called when a block and its transactions are downloaded."""
         # Note: Any vertex and block could have already been added by another concurrent syncing peer.
         try:
-            for tx in vertex_list:
-                if not self.tx_storage.transaction_exists(tx.hash):
-                    self.vertex_handler.on_new_vertex(tx, fails_silently=False)
-                yield deferLater(self.reactor, 0, lambda: None)
-
-            if not self.tx_storage.transaction_exists(blk.hash):
-                self.vertex_handler.on_new_vertex(blk, fails_silently=False)
+            yield self.vertex_handler.on_new_block(blk, deps=vertex_list)
         except InvalidNewTransaction:
             self.protocol.send_error_and_close_connection('invalid vertex received')
 
@@ -1038,6 +1032,7 @@ def handle_transaction(self, payload: str) -> None:
         tx.storage = self.tx_storage
 
         assert self._tx_streaming_client is not None
+        assert isinstance(tx, Transaction)
         self._tx_streaming_client.handle_transaction(tx)
 
     @inlineCallbacks
@@ -1166,17 +1161,18 @@ def handle_data(self, payload: str) -> None:
             # XXX: maybe we could add a hash blacklist and punish peers propagating known bad txs
             self.tx_storage.compare_bytes_with_local_tx(tx)
             return
-        else:
-            # If we have not requested the data, it is a new transaction being propagated
-            # in the network, thus, we propagate it as well.
-            if self.tx_storage.can_validate_full(tx):
-                self.log.debug('tx received in real time from peer', tx=tx.hash_hex, peer=self.protocol.get_peer_id())
-                try:
-                    success = self.vertex_handler.on_new_vertex(tx, fails_silently=False)
-                    if success:
-                        self.protocol.connections.send_tx_to_peers(tx)
-                except InvalidNewTransaction:
-                    self.protocol.send_error_and_close_connection('invalid vertex received')
-            else:
-                self.log.debug('skipping tx received in real time from peer',
-                               tx=tx.hash_hex, peer=self.protocol.get_peer_id())
+
+        # Unsolicited vertices must be fully validated.
+        if not self.tx_storage.can_validate_full(tx):
+            self.log.debug('skipping tx received in real time from peer',
+                           tx=tx.hash_hex, peer=self.protocol.get_peer_id())
+            return
+
+        # Finally, it is either an unsolicited new transaction or block.
+        self.log.debug('tx received in real time from peer', tx=tx.hash_hex, peer=self.protocol.get_peer_id())
+        try:
+            success = self.vertex_handler.on_new_relayed_vertex(tx)
+            if success:
+                self.protocol.connections.send_tx_to_peers(tx)
+        except InvalidNewTransaction:
+            self.protocol.send_error_and_close_connection('invalid vertex received')

hathor/p2p/sync_v2/blockchain_streaming_client.py

@@ -126,7 +126,7 @@ def handle_blocks(self, blk: Block) -> None:
 
         if self.tx_storage.can_validate_full(blk):
             try:
-                self.vertex_handler.on_new_vertex(blk, fails_silently=False)
+                self.vertex_handler.on_new_block(blk, deps=[])
             except HathorError:
                 self.fails(InvalidVertexError(blk.hash.hex()))
                 return

hathor/p2p/sync_v2/mempool.py

@@ -19,7 +19,7 @@
 from twisted.internet.defer import Deferred, inlineCallbacks
 
 from hathor.exception import InvalidNewTransaction
-from hathor.transaction import BaseTransaction
+from hathor.transaction import Transaction
 
 if TYPE_CHECKING:
     from hathor.p2p.sync_v2.agent import NodeBlockSync
@@ -95,7 +95,7 @@ def _unsafe_run(self) -> Generator[Deferred, Any, bool]:
         while self.missing_tips:
             self.log.debug('We have missing tips! Let\'s start!', missing_tips=[x.hex() for x in self.missing_tips])
             tx_id = next(iter(self.missing_tips))
-            tx: BaseTransaction = yield self.sync_agent.get_tx(tx_id)
+            tx: Transaction = yield self.sync_agent.get_tx(tx_id)
             # Stack used by the DFS in the dependencies.
             # We use a deque for performance reasons.
             self.log.debug('start mempool DSF', tx=tx.hash_hex)
@@ -106,7 +106,7 @@ def _unsafe_run(self) -> Generator[Deferred, Any, bool]:
         return False
 
     @inlineCallbacks
-    def _dfs(self, stack: deque[BaseTransaction]) -> Generator[Deferred, Any, None]:
+    def _dfs(self, stack: deque[Transaction]) -> Generator[Deferred, Any, None]:
         """DFS method."""
         while stack:
             tx = stack[-1]
@@ -123,7 +123,7 @@ def _dfs(self, stack: deque[BaseTransaction]) -> Generator[Deferred, Any, None]:
                 if len(stack) > self.MAX_STACK_LENGTH:
                     stack.popleft()
 
-    def _next_missing_dep(self, tx: BaseTransaction) -> Optional[bytes]:
+    def _next_missing_dep(self, tx: Transaction) -> Optional[bytes]:
         """Get the first missing dependency found of tx."""
         assert not tx.is_block
         for txin in tx.inputs:
@@ -134,13 +134,13 @@ def _next_missing_dep(self, tx: BaseTransaction) -> Optional[bytes]:
                 return parent
         return None
 
-    def _add_tx(self, tx: BaseTransaction) -> None:
+    def _add_tx(self, tx: Transaction) -> None:
         """Add tx to the DAG."""
         self.missing_tips.discard(tx.hash)
         if self.tx_storage.transaction_exists(tx.hash):
             return
         try:
-            success = self.vertex_handler.on_new_vertex(tx, fails_silently=False)
+            success = self.vertex_handler.on_new_mempool_transaction(tx)
             if success:
                 self.sync_agent.protocol.connections.send_tx_to_peers(tx)
         except InvalidNewTransaction:

hathor/p2p/sync_v2/transaction_streaming_client.py

@@ -25,7 +25,7 @@
     UnexpectedVertex,
 )
 from hathor.p2p.sync_v2.streamers import StreamEnd
-from hathor.transaction import BaseTransaction
+from hathor.transaction import BaseTransaction, Transaction
 from hathor.transaction.exceptions import HathorError, TxValidationError
 from hathor.types import VertexId
 
@@ -66,7 +66,7 @@ def __init__(self,
         self._tx_max_quantity = limit
 
         # Queue of transactions waiting to be processed.
-        self._queue: deque[BaseTransaction] = deque()
+        self._queue: deque[Transaction] = deque()
 
         # Keeps the response code if the streaming has ended.
         self._response_code: Optional[StreamEnd] = None
@@ -79,7 +79,7 @@ def __init__(self,
 
         # In-memory database of transactions already received but still
         # waiting for dependencies.
-        self._db: dict[VertexId, BaseTransaction] = {}
+        self._db: dict[VertexId, Transaction] = {}
         self._existing_deps: set[VertexId] = set()
 
         self._prepare_block(self.partial_blocks[0])
@@ -103,7 +103,7 @@ def fails(self, reason: 'StreamingError') -> None:
             return
         self._deferred.errback(reason)
 
-    def handle_transaction(self, tx: BaseTransaction) -> None:
+    def handle_transaction(self, tx: Transaction) -> None:
         """This method is called by the sync agent when a TRANSACTION message is received."""
         if self._deferred.called:
             return
@@ -147,7 +147,7 @@ def process_queue(self) -> Generator[Any, Any, None]:
         self.reactor.callLater(0, self.process_queue)
 
     @inlineCallbacks
-    def _process_transaction(self, tx: BaseTransaction) -> Generator[Any, Any, None]:
+    def _process_transaction(self, tx: Transaction) -> Generator[Any, Any, None]:
         """Process transaction."""
 
         # Run basic verification.
@@ -177,6 +177,7 @@ def _process_transaction(self, tx: BaseTransaction) -> Generator[Any, Any, None]
 
         self._update_dependencies(tx)
 
+        assert isinstance(tx, Transaction)
         self._db[tx.hash] = tx
 
         if not self._waiting_for:
@@ -191,9 +192,9 @@ def _process_transaction(self, tx: BaseTransaction) -> Generator[Any, Any, None]
         if self._tx_received % 100 == 0:
             self.log.debug('tx streaming in progress', txs_received=self._tx_received)
 
-    def _update_dependencies(self, tx: BaseTransaction) -> None:
+    def _update_dependencies(self, vertex: BaseTransaction) -> None:
         """Update _existing_deps and _waiting_for with the dependencies."""
-        for dep in tx.get_all_dependencies():
+        for dep in vertex.get_all_dependencies():
             if self.tx_storage.transaction_exists(dep) or dep in self._db:
                 self._existing_deps.add(dep)
             else:

hathor/simulator/miner/geometric_miner.py

@@ -91,7 +91,7 @@ def _schedule_next_block(self):
             self._block.nonce = self._rng.getrandbits(32)
             self._block.update_hash()
             self.log.debug('randomized step: found new block', hash=self._block.hash_hex, nonce=self._block.nonce)
-            self._manager.propagate_tx(self._block, fails_silently=False)
+            self._manager.propagate_tx(self._block)
             self._blocks_found += 1
             self._blocks_before_pause -= 1
             self._block = None