review changes

Author: glevco

hathor/builder/builder.py

@@ -312,12 +312,7 @@ def _get_peer_id(self) -> PeerId:
 
     def _get_or_create_execution_manager(self) -> ExecutionManager:
         if self._execution_manager is None:
-            tx_storage = self._get_or_create_tx_storage()
-            event_manager = self._get_or_create_event_manager()
-            self._execution_manager = ExecutionManager(
-                tx_storage=tx_storage,
-                event_manager=event_manager,
-            )
+            self._execution_manager = ExecutionManager()
 
         return self._execution_manager
 
@@ -453,7 +448,8 @@ def _get_or_create_event_manager(self) -> EventManager:
                 reactor=reactor,
                 pubsub=self._get_or_create_pubsub(),
                 event_storage=storage,
-                event_ws_factory=factory
+                event_ws_factory=factory,
+                execution_manager=self._get_or_create_execution_manager()
             )
 
         return self._event_manager

hathor/builder/cli_builder.py

@@ -172,11 +172,14 @@ def create_manager(self, reactor: Reactor) -> HathorManager:
                 event_storage=event_storage
             )
 
+        execution_manager = ExecutionManager()
+
         event_manager = EventManager(
             event_storage=event_storage,
             event_ws_factory=self.event_ws_factory,
             pubsub=pubsub,
-            reactor=reactor
+            reactor=reactor,
+            execution_manager=execution_manager,
         )
 
         if self._args.wallet_index and tx_storage.indexes is not None:
@@ -196,10 +199,6 @@ def create_manager(self, reactor: Reactor) -> HathorManager:
             full_verification = True
 
         soft_voided_tx_ids = set(settings.SOFT_VOIDED_TX_IDS)
-        execution_manager = ExecutionManager(
-            tx_storage=tx_storage,
-            event_manager=event_manager,
-        )
         consensus_algorithm = ConsensusAlgorithm(
             soft_voided_tx_ids,
             pubsub=pubsub,
@@ -276,7 +275,8 @@ def create_manager(self, reactor: Reactor) -> HathorManager:
             feature_service=self.feature_service,
             bit_signaling_service=bit_signaling_service,
             verification_service=verification_service,
-            cpu_mining_service=cpu_mining_service
+            cpu_mining_service=cpu_mining_service,
+            execution_manager=execution_manager,
         )
 
         if self._args.x_ipython_kernel:

hathor/consensus/consensus.py

@@ -83,11 +83,13 @@ def update(self, base: BaseTransaction) -> None:
         assert meta.validation.is_valid()
         try:
             self._unsafe_update(base)
-        except BaseException:
+        except BaseException as e:
             meta.add_voided_by(self._settings.CONSENSUS_FAIL_ID)
             assert base.storage is not None
             base.storage.save_transaction(base, only_metadata=True)
-            self._execution_manager.crash_and_exit(reason=f'Consensus update failed for tx {base.hash_hex}')
+            message = f'Consensus update failed for tx {base.hash_hex}'
+            self.log.critical(f'{message}. Exception: {repr(e)}', exc_info=True)
+            self._execution_manager.crash_and_exit(reason=message)
 
     def _unsafe_update(self, base: BaseTransaction) -> None:
         """Run a consensus update with its own context, indexes will be updated accordingly."""

hathor/event/event_manager.py

@@ -22,6 +22,7 @@
 from hathor.event.model.node_state import NodeState
 from hathor.event.storage import EventStorage
 from hathor.event.websocket import EventWebsocketFactory
+from hathor.execution_manager import ExecutionManager
 from hathor.pubsub import EventArguments, HathorEvents, PubSubManager
 from hathor.reactor import ReactorProtocol as Reactor
 from hathor.transaction import BaseTransaction
@@ -70,6 +71,7 @@ def __init__(
         event_storage: EventStorage,
         pubsub: PubSubManager,
         reactor: Reactor,
+        execution_manager: ExecutionManager,
         event_ws_factory: Optional[EventWebsocketFactory] = None,
     ) -> None:
         self.log = logger.new()
@@ -78,13 +80,15 @@ def __init__(
         self._event_storage = event_storage
         self._event_ws_factory = event_ws_factory
         self._pubsub = pubsub
+        self._execution_manager = execution_manager
 
     def start(self, peer_id: str) -> None:
         """Starts the EventManager."""
         assert self._is_running is False, 'Cannot start, EventManager is already running'
         assert self._event_ws_factory is not None, 'Cannot start, EventWebsocketFactory is not set'
         assert self.get_event_queue_state() is True, 'Cannot start, event queue feature is disabled'
 
+        self._execution_manager.register_on_crash_callback(self.on_full_node_crash)
         self._previous_node_state = self._event_storage.get_node_state()
 
         if self._should_reload_events():
@@ -153,7 +157,7 @@ def load_finished(self) -> None:
         )
         self._event_storage.save_node_state(NodeState.SYNC)
 
-    def full_node_crashed(self) -> None:
+    def on_full_node_crash(self) -> None:
         if not self._is_running:
             return
 

hathor/execution_manager.py

@@ -13,24 +13,34 @@
 #  limitations under the License.
 
 import os
-from typing import NoReturn
+from typing import Callable, NoReturn
 
 from structlog import get_logger
 
-from hathor.event import EventManager
-from hathor.transaction.storage import TransactionStorage
-
 logger = get_logger()
 
 
 class ExecutionManager:
     """Class to manage actions related to full node execution."""
-    __slots__ = ('_log', '_tx_storage', '_event_manager')
+    __slots__ = ('_log', '_on_crash_callbacks')
 
-    def __init__(self, *, tx_storage: TransactionStorage, event_manager: EventManager) -> None:
+    def __init__(self) -> None:
         self._log = logger.new()
-        self._tx_storage = tx_storage
-        self._event_manager = event_manager
+        self._on_crash_callbacks: list[tuple[int, Callable[[], None]]] = []
+
+    def register_on_crash_callback(self, callback: Callable[[], None], *, priority: int = 0) -> None:
+        """Register a callback to be executed before the full node exits."""
+        self._on_crash_callbacks.append((priority, callback))
+
+    def _run_on_crash_callbacks(self) -> None:
+        """Run all registered on crash callbacks."""
+        callbacks = sorted(self._on_crash_callbacks, reverse=True)
+
+        for _, callback in callbacks:
+            try:
+                callback()
+            except BaseException as e:
+                self._log.critical(f'Failed execution of on_crash callback "{callback}". Exception: {repr(e)}')
 
     def crash_and_exit(self, *, reason: str) -> NoReturn:
         """
@@ -39,8 +49,7 @@ def crash_and_exit(self, *, reason: str) -> NoReturn:
         corrupted, and requiring manual intervention. In other words, a restart with a clean database (from scratch
         or a snapshot) will be required.
         """
-        self._tx_storage.full_node_crashed()
-        self._event_manager.full_node_crashed()
+        self._run_on_crash_callbacks()
         self._log.critical(
             'Critical failure occurred, causing the full node to halt execution. Manual intervention is required.',
             reason=reason,

hathor/manager.py

@@ -40,6 +40,7 @@
     RewardLockedError,
     SpendingVoidedError,
 )
+from hathor.execution_manager import ExecutionManager
 from hathor.feature_activation.bit_signaling_service import BitSignalingService
 from hathor.feature_activation.feature import Feature
 from hathor.feature_activation.feature_service import FeatureService
@@ -102,6 +103,7 @@ def __init__(self,
                  verification_service: VerificationService,
                  cpu_mining_service: CpuMiningService,
                  network: str,
+                 execution_manager: ExecutionManager,
                  hostname: Optional[str] = None,
                  wallet: Optional[BaseWallet] = None,
                  capabilities: Optional[list[str]] = None,
@@ -127,6 +129,7 @@ def __init__(self,
                 'Either enable it, or use the reset-event-queue CLI command to remove all event-related data'
             )
 
+        self._execution_manager = execution_manager
         self._settings = settings
         self.daa = daa
         self._cmd_path: Optional[str] = None
@@ -322,7 +325,7 @@ def start(self) -> None:
             self.stratum_factory.start()
 
         # Start running
-        self.tx_storage.start_running_manager()
+        self.tx_storage.start_running_manager(self._execution_manager)
 
     def stop(self) -> Deferred:
         if not self.is_started:

hathor/transaction/storage/transaction_storage.py

@@ -24,6 +24,7 @@
 from structlog import get_logger
 
 from hathor.conf.get_settings import get_settings
+from hathor.execution_manager import ExecutionManager
 from hathor.indexes import IndexesManager
 from hathor.indexes.height_index import HeightInfo
 from hathor.profiler import get_cpu_profiler
@@ -966,9 +967,10 @@ def is_running_full_verification(self) -> bool:
         """
         return self.get_value(self._running_full_verification_attribute) == '1'
 
-    def start_running_manager(self) -> None:
+    def start_running_manager(self, execution_manager: ExecutionManager) -> None:
         """ Save on storage that manager is running
         """
+        execution_manager.register_on_crash_callback(self.on_full_node_crash)
         self.add_value(self._manager_running_attribute, '1')
 
     def stop_running_manager(self) -> None:
@@ -981,7 +983,7 @@ def is_running_manager(self) -> bool:
         """
         return self.get_value(self._manager_running_attribute) == '1'
 
-    def full_node_crashed(self) -> None:
+    def on_full_node_crash(self) -> None:
         """Save on storage that the full node crashed and cannot be recovered."""
         self.add_value(self._full_node_crashed_attribute, '1')
 

tests/execution_manager/test_execution_manager.py

@@ -15,24 +15,25 @@
 import os
 from unittest.mock import Mock, patch
 
-from hathor.event import EventManager
 from hathor.execution_manager import ExecutionManager
-from hathor.transaction.storage import TransactionStorage
 
 
 def test_crash_and_exit() -> None:
-    tx_storage_mock = Mock(spec_set=TransactionStorage)
-    event_manager_mock = Mock(spec_set=EventManager)
+    def callback() -> None:
+        pass
+
+    callback_wrapped = Mock(wraps=callback)
     log_mock = Mock()
-    manager = ExecutionManager(tx_storage=tx_storage_mock, event_manager=event_manager_mock)
+    manager = ExecutionManager()
     manager._log = log_mock
     reason = 'some critical failure'
 
+    manager.register_on_crash_callback(callback_wrapped)
+
     with patch.object(os, '_exit') as exit_mock:
         manager.crash_and_exit(reason=reason)
 
-    tx_storage_mock.full_node_crashed.assert_called_once()
-    event_manager_mock.full_node_crashed.assert_called_once()
+    callback_wrapped.assert_called_once()
     log_mock.critical.assert_called_once_with(
         'Critical failure occurred, causing the full node to halt execution. Manual intervention is required.',
         reason=reason,