feat(unrecoverable-error): implement halting of full node execution

Author: glevco

hathor/builder/builder.py

@@ -24,6 +24,7 @@
 from hathor.event import EventManager
 from hathor.event.storage import EventMemoryStorage, EventRocksDBStorage, EventStorage
 from hathor.event.websocket import EventWebsocketFactory
+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
@@ -137,6 +138,8 @@ def __init__(self) -> None:
 
         self._soft_voided_tx_ids: Optional[set[bytes]] = None
 
+        self._execution_manager: ExecutionManager | None = None
+
     def build(self) -> BuildArtifacts:
         if self.artifacts is not None:
             raise ValueError('cannot call build twice')
@@ -150,15 +153,17 @@ def build(self) -> BuildArtifacts:
 
         peer_id = self._get_peer_id()
 
+        indexes = self._get_or_create_indexes_manager()
+        tx_storage = self._get_or_create_tx_storage(indexes)
+
+        execution_manager = self._get_or_create_execution_manager(tx_storage)
         soft_voided_tx_ids = self._get_soft_voided_tx_ids()
-        consensus_algorithm = ConsensusAlgorithm(soft_voided_tx_ids, pubsub)
+        consensus_algorithm = ConsensusAlgorithm(soft_voided_tx_ids, pubsub, execution_manager=execution_manager)
 
         p2p_manager = self._get_p2p_manager()
 
         wallet = self._get_or_create_wallet()
         event_manager = self._get_or_create_event_manager()
-        indexes = self._get_or_create_indexes_manager()
-        tx_storage = self._get_or_create_tx_storage(indexes)
         feature_service = self._get_or_create_feature_service(tx_storage)
         bit_signaling_service = self._get_or_create_bit_signaling_service(tx_storage)
         verification_service = self._get_or_create_verification_service()
@@ -287,6 +292,16 @@ def _get_peer_id(self) -> PeerId:
             return self._peer_id
         raise ValueError('peer_id not set')
 
+    def _get_or_create_execution_manager(self, tx_storage: TransactionStorage) -> ExecutionManager:
+        if self._execution_manager is None:
+            event_manager = self._get_or_create_event_manager()
+            self._execution_manager = ExecutionManager(
+                tx_storage=tx_storage,
+                event_manager=event_manager,
+            )
+
+        return self._execution_manager
+
     def _get_or_create_pubsub(self) -> PubSubManager:
         if self._pubsub is None:
             self._pubsub = PubSubManager(self._get_reactor())

hathor/builder/cli_builder.py

@@ -25,6 +25,7 @@
 from hathor.consensus import ConsensusAlgorithm
 from hathor.event import EventManager
 from hathor.exception import BuilderError
+from hathor.execution_manager import ExecutionManager
 from hathor.feature_activation.bit_signaling_service import BitSignalingService
 from hathor.feature_activation.feature_service import FeatureService
 from hathor.indexes import IndexesManager, MemoryIndexesManager, RocksDBIndexesManager
@@ -183,7 +184,15 @@ def create_manager(self, reactor: Reactor) -> HathorManager:
             full_verification = True
 
         soft_voided_tx_ids = set(settings.SOFT_VOIDED_TX_IDS)
-        consensus_algorithm = ConsensusAlgorithm(soft_voided_tx_ids, pubsub=pubsub)
+        execution_manager = ExecutionManager(
+            tx_storage=tx_storage,
+            event_manager=event_manager,
+        )
+        consensus_algorithm = ConsensusAlgorithm(
+            soft_voided_tx_ids,
+            pubsub=pubsub,
+            execution_manager=execution_manager
+        )
 
         if self._args.x_enable_event_queue:
             self.log.info('--x-enable-event-queue flag provided. '

hathor/consensus/consensus.py

@@ -18,6 +18,7 @@
 from hathor.consensus.block_consensus import BlockConsensusAlgorithmFactory
 from hathor.consensus.context import ConsensusAlgorithmContext
 from hathor.consensus.transaction_consensus import TransactionConsensusAlgorithmFactory
+from hathor.execution_manager import ExecutionManager
 from hathor.profiler import get_cpu_profiler
 from hathor.pubsub import HathorEvents, PubSubManager
 from hathor.transaction import BaseTransaction
@@ -55,13 +56,20 @@ class ConsensusAlgorithm:
     b0 will not be propagated to the voided_by of b1, b2, and b3.
     """
 
-    def __init__(self, soft_voided_tx_ids: set[bytes], pubsub: PubSubManager) -> None:
+    def __init__(
+        self,
+        soft_voided_tx_ids: set[bytes],
+        pubsub: PubSubManager,
+        *,
+        execution_manager: ExecutionManager
+    ) -> None:
         self._settings = get_settings()
         self.log = logger.new()
         self._pubsub = pubsub
         self.soft_voided_tx_ids = frozenset(soft_voided_tx_ids)
         self.block_algorithm_factory = BlockConsensusAlgorithmFactory()
         self.transaction_algorithm_factory = TransactionConsensusAlgorithmFactory()
+        self._execution_manager = execution_manager
 
     def create_context(self) -> ConsensusAlgorithmContext:
         """Handy method to create a context that can be used to access block and transaction algorithms."""
@@ -75,11 +83,11 @@ def update(self, base: BaseTransaction) -> None:
         assert meta.validation.is_valid()
         try:
             self._unsafe_update(base)
-        except Exception:
+        except BaseException:
             meta.add_voided_by(self._settings.CONSENSUS_FAIL_ID)
             assert base.storage is not None
             base.storage.save_transaction(base, only_metadata=True)
-            raise
+            self._execution_manager.crash_and_exit(reason=f'Consensus update failed for tx {base.hash_hex}')
 
     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

@@ -132,7 +132,7 @@ def _subscribe_events(self) -> None:
         for event in _SUBSCRIBE_EVENTS:
             self._pubsub.subscribe(event, self._handle_hathor_event)
 
-    def load_started(self):
+    def load_started(self) -> None:
         if not self._is_running:
             return
 
@@ -142,7 +142,7 @@ def load_started(self):
         )
         self._event_storage.save_node_state(NodeState.LOAD)
 
-    def load_finished(self):
+    def load_finished(self) -> None:
         if not self._is_running:
             return
 
@@ -152,6 +152,15 @@ def load_finished(self):
         )
         self._event_storage.save_node_state(NodeState.SYNC)
 
+    def full_node_crashed(self) -> None:
+        if not self._is_running:
+            return
+
+        self._handle_event(
+            event_type=EventType.FULL_NODE_CRASHED,
+            event_args=EventArguments(),
+        )
+
     def _handle_hathor_event(self, hathor_event: HathorEvents, event_args: EventArguments) -> None:
         """Handles a PubSub 'HathorEvents' event."""
         event_type = EventType.from_hathor_event(hathor_event)

hathor/event/model/event_type.py

@@ -25,6 +25,7 @@ class EventType(Enum):
     REORG_STARTED = 'REORG_STARTED'
     REORG_FINISHED = 'REORG_FINISHED'
     VERTEX_METADATA_CHANGED = 'VERTEX_METADATA_CHANGED'
+    FULL_NODE_CRASHED = 'FULL_NODE_CRASHED'
 
     @classmethod
     def from_hathor_event(cls, hathor_event: HathorEvents) -> 'EventType':
@@ -53,4 +54,5 @@ def data_type(self) -> type[BaseEventData]:
     EventType.REORG_STARTED: ReorgData,
     EventType.REORG_FINISHED: EmptyData,
     EventType.VERTEX_METADATA_CHANGED: TxData,
+    EventType.FULL_NODE_CRASHED: EmptyData,
 }

hathor/execution_manager.py

@@ -0,0 +1,51 @@
+#  Copyright 2023 Hathor Labs
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+import os
+from typing import 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')
+
+    def __init__(self, *, tx_storage: TransactionStorage, event_manager: EventManager) -> None:
+        self._log = logger.new()
+        self._tx_storage = tx_storage
+        self._event_manager = event_manager
+
+    def crash_and_exit(self, *, reason: str) -> NoReturn:
+        """
+        Calling this function is a very extreme thing to do, so be careful. It should only be called when a
+        critical, unrecoverable failure happens. It crashes and exits the full node, rendering the database
+        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._log.critical(
+            'Critical failure occurred, causing the full node to halt execution. Manual intervention is required.',
+            reason=reason,
+            exc_info=True
+        )
+        # We use os._exit() instead of sys.exit() or any other approaches because this is the only one Twisted
+        # doesn't catch.
+        os._exit(-1)

hathor/manager.py

@@ -238,6 +238,15 @@ def start(self) -> None:
         self.is_started = True
 
         self.log.info('start manager', network=self.network)
+
+        if self.tx_storage.is_full_node_crashed():
+            self.log.error(
+                'Error initializing node. The last time you executed your full node it wasn\'t stopped correctly. '
+                'The storage is not reliable anymore and, because of that, you must remove your storage and do a '
+                'full sync (either from scratch or from a snapshot).'
+            )
+            sys.exit(-1)
+
         # If it's a full verification, we save on the storage that we are starting it
         # this is required because if we stop the initilization in the middle, the metadata
         # saved on the storage is not reliable anymore, only if we finish it
@@ -1011,13 +1020,7 @@ def on_new_tx(self, tx: BaseTransaction, *, conn: Optional[HathorProtocol] = Non
         tx.update_initial_metadata(save=False)
         self.tx_storage.save_transaction(tx)
         self.tx_storage.add_to_indexes(tx)
-        try:
-            self.consensus_algorithm.update(tx)
-        except HathorError as e:
-            if not fails_silently:
-                raise InvalidNewTransaction('consensus update failed') from e
-            self.log.warn('on_new_tx(): consensus update failed', tx=tx.hash_hex, exc_info=True)
-            return False
+        self.consensus_algorithm.update(tx)
 
         assert self.verification_service.validate_full(
             tx,

hathor/transaction/storage/transaction_storage.py

@@ -75,6 +75,9 @@ class TransactionStorage(ABC):
     # Key storage attribute to save if the manager is running
     _manager_running_attribute: str = 'manager_running'
 
+    # Key storage attribute to save if the full node crashed
+    _full_node_crashed_attribute: str = 'full_node_crashed'
+
     # Ket storage attribute to save the last time the node started
     _last_start_attribute: str = 'last_start'
 
@@ -970,6 +973,14 @@ def is_running_manager(self) -> bool:
         """
         return self.get_value(self._manager_running_attribute) == '1'
 
+    def full_node_crashed(self) -> None:
+        """Save on storage that the full node crashed and cannot be recovered."""
+        self.add_value(self._full_node_crashed_attribute, '1')
+
+    def is_full_node_crashed(self) -> bool:
+        """Return whether the full node was crashed."""
+        return self.get_value(self._full_node_crashed_attribute) == '1'
+
     def get_last_started_at(self) -> int:
         """ Return the timestamp when the database was last started.
         """

tests/consensus/test_consensus.py

@@ -1,6 +1,7 @@
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, Mock
 
 from hathor.conf import HathorSettings
+from hathor.execution_manager import ExecutionManager
 from hathor.transaction.storage import TransactionMemoryStorage
 from tests import unittest
 from tests.utils import (
@@ -39,10 +40,15 @@ def test_unhandled_exception(self):
         class MyError(Exception):
             pass
 
+        execution_manager_mock = Mock(spec_set=ExecutionManager)
+        manager.consensus_algorithm._execution_manager = execution_manager_mock
         manager.consensus_algorithm._unsafe_update = MagicMock(side_effect=MyError)
 
-        with self.assertRaises(MyError):
-            manager.propagate_tx(tx, fails_silently=False)
+        manager.propagate_tx(tx, fails_silently=False)
+
+        execution_manager_mock.crash_and_exit.assert_called_once_with(
+            reason=f"Consensus update failed for tx {tx.hash_hex}"
+        )
 
         tx2 = manager.tx_storage.get_transaction(tx.hash)
         meta2 = tx2.get_metadata()

tests/execution_manager/__init__.py

@@ -0,0 +1,42 @@
+#  Copyright 2023 Hathor Labs
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#  http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+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)
+    log_mock = Mock()
+    manager = ExecutionManager(tx_storage=tx_storage_mock, event_manager=event_manager_mock)
+    manager._log = log_mock
+    reason = 'some critical failure'
+
+    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()
+    log_mock.critical.assert_called_once_with(
+        'Critical failure occurred, causing the full node to halt execution. Manual intervention is required.',
+        reason=reason,
+        exc_info=True
+    )
+
+    exit_mock.assert_called_once_with(-1)