feat(event-queue): add new VERTEX_REMOVED event

hathor/consensus/consensus.py

@@ -122,21 +122,20 @@ def _unsafe_update(self, base: BaseTransaction) -> None:
             raise NotImplementedError
 
         new_best_height, new_best_tip = storage.indexes.height.get_height_tip()
+        txs_to_remove: list[BaseTransaction] = []
         if new_best_height < best_height:
             self.log.warn('height decreased, re-checking mempool', prev_height=best_height, new_height=new_best_height,
                           prev_block_tip=best_tip.hex(), new_block_tip=new_best_tip.hex())
             # XXX: this method will mark as INVALID all transactions in the mempool that became invalid because of a
             #      reward lock
-            to_remove = storage.compute_transactions_that_became_invalid(new_best_height)
-            if to_remove:
+            txs_to_remove = storage.compute_transactions_that_became_invalid(new_best_height)
+            if txs_to_remove:
                 self.log.warn('some transactions on the mempool became invalid and will be removed',
-                              count=len(to_remove))
-                # XXX: because transactions in `to_remove` are marked as invalid, we need this context to be able to
-                #      remove them
+                              count=len(txs_to_remove))
+                # XXX: because transactions in `txs_to_remove` are marked as invalid, we need this context to be
+                # able to remove them
                 with storage.allow_invalid_context():
-                    self._remove_transactions(to_remove, storage, context)
-                for tx_removed in to_remove:
-                    context.pubsub.publish(HathorEvents.CONSENSUS_TX_REMOVED, tx_hash=tx_removed.hash)
+                    self._remove_transactions(txs_to_remove, storage, context)
 
         # emit the reorg started event if needed
         if context.reorg_common_block is not None:
@@ -158,6 +157,10 @@ def _unsafe_update(self, base: BaseTransaction) -> None:
             tx_affected.storage.indexes.update(tx_affected)
             context.pubsub.publish(HathorEvents.CONSENSUS_TX_UPDATE, tx=tx_affected)
 
+        # And emit events for txs that were removed
+        for tx_removed in txs_to_remove:
+            context.pubsub.publish(HathorEvents.CONSENSUS_TX_REMOVED, vertex_id=tx_removed.hash)
+
         # and also emit the reorg finished event if needed
         if context.reorg_common_block is not None:
             context.pubsub.publish(HathorEvents.REORG_FINISHED)

hathor/event/event_manager.py

@@ -46,6 +46,7 @@
     HathorEvents.REORG_STARTED,
     HathorEvents.REORG_FINISHED,
     HathorEvents.CONSENSUS_TX_UPDATE,
+    HathorEvents.CONSENSUS_TX_REMOVED,
 ]
 
 

hathor/event/model/event_data.py

@@ -12,9 +12,10 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from typing import Optional, Union, cast
+from typing import Optional, TypeAlias, Union, cast
 
 from pydantic import Extra, validator
+from typing_extensions import Self
 
 from hathor.pubsub import EventArguments
 from hathor.utils.pydantic import BaseModel
@@ -137,6 +138,14 @@ def from_event_arguments(cls, args: EventArguments) -> 'TxData':
         return cls(**tx_json)
 
 
+class VertexIdData(BaseEventData):
+    vertex_id: str
+
+    @classmethod
+    def from_event_arguments(cls, args: EventArguments) -> Self:
+        return cls(vertex_id=args.vertex_id.hex())
+
+
 class ReorgData(BaseEventData):
     """Class that represents reorg data on an event."""
     reorg_size: int
@@ -155,4 +164,4 @@ def from_event_arguments(cls, args: EventArguments) -> 'ReorgData':
 
 
 # Union type to encompass BaseEventData polymorphism
-EventData = Union[EmptyData, TxData, ReorgData]
+EventData: TypeAlias = EmptyData | TxData | ReorgData | VertexIdData

hathor/event/model/event_type.py

@@ -14,7 +14,7 @@
 
 from enum import Enum
 
-from hathor.event.model.event_data import BaseEventData, EmptyData, ReorgData, TxData
+from hathor.event.model.event_data import BaseEventData, EmptyData, ReorgData, TxData, VertexIdData
 from hathor.pubsub import HathorEvents
 
 
@@ -25,6 +25,7 @@ class EventType(Enum):
     REORG_STARTED = 'REORG_STARTED'
     REORG_FINISHED = 'REORG_FINISHED'
     VERTEX_METADATA_CHANGED = 'VERTEX_METADATA_CHANGED'
+    VERTEX_REMOVED = 'VERTEX_REMOVED'
     FULL_NODE_CRASHED = 'FULL_NODE_CRASHED'
 
     @classmethod
@@ -44,7 +45,8 @@ def data_type(self) -> type[BaseEventData]:
     HathorEvents.NETWORK_NEW_TX_ACCEPTED: EventType.NEW_VERTEX_ACCEPTED,
     HathorEvents.REORG_STARTED: EventType.REORG_STARTED,
     HathorEvents.REORG_FINISHED: EventType.REORG_FINISHED,
-    HathorEvents.CONSENSUS_TX_UPDATE: EventType.VERTEX_METADATA_CHANGED
+    HathorEvents.CONSENSUS_TX_UPDATE: EventType.VERTEX_METADATA_CHANGED,
+    HathorEvents.CONSENSUS_TX_REMOVED: EventType.VERTEX_REMOVED
 }
 
 _EVENT_TYPE_TO_EVENT_DATA: dict[EventType, type[BaseEventData]] = {
@@ -54,5 +56,6 @@ def data_type(self) -> type[BaseEventData]:
     EventType.REORG_STARTED: ReorgData,
     EventType.REORG_FINISHED: EmptyData,
     EventType.VERTEX_METADATA_CHANGED: TxData,
+    EventType.VERTEX_REMOVED: VertexIdData,
     EventType.FULL_NODE_CRASHED: EmptyData,
 }

hathor/pubsub.py

@@ -21,6 +21,7 @@
 from twisted.python.threadable import isInIOThread
 
 from hathor.reactor import ReactorProtocol as Reactor
+from hathor.types import VertexId
 from hathor.utils.zope import verified_cast
 
 if TYPE_CHECKING:
@@ -145,6 +146,7 @@ class EventArguments:
 
     # XXX: add these as needed, these attributes don't always exist, but when they do these are their types
     tx: 'BaseTransaction'
+    vertex_id: VertexId
     reorg_size: int
     old_best_block: 'Block'
     new_best_block: 'Block'