Internal: TX processing is now event-based

Problem: the pending TX processor uses a polling loop to determine which
transactions must be processed. This leads to some latency when adding a
new transaction when there is no currently pending transaction as the
task sleeps while waiting for new transactions.

Solution: make the TX processor event-based by using a RabbitMQ exchange
+ message queue. Upon receiving a new transaction from on-chain, we now
publish a RabbitMQ message containing the hash of the transaction. The
processor can then find the transaction in the DB.

We still store an entry in the pending TXs table for monitoring purposes.

src/aleph/chains/bsc.py

@@ -1,21 +1,23 @@
 from aleph_message.models import Chain
 from configmanager import Config
 
-from aleph.chains.chain_data_service import ChainDataService
 from aleph.chains.abc import ChainReader
+from aleph.chains.chain_data_service import PendingTxPublisher
 from aleph.chains.indexer_reader import AlephIndexerReader
 from aleph.types.chain_sync import ChainEventType
 from aleph.types.db_session import DbSessionFactory
 
 
 class BscConnector(ChainReader):
     def __init__(
-        self, session_factory: DbSessionFactory, chain_data_service: ChainDataService
+        self,
+        session_factory: DbSessionFactory,
+        pending_tx_publisher: PendingTxPublisher,
     ):
         self.indexer_reader = AlephIndexerReader(
             chain=Chain.BSC,
             session_factory=session_factory,
-            chain_data_service=chain_data_service,
+            pending_tx_publisher=pending_tx_publisher,
         )
 
     async def fetcher(self, config: Config):

src/aleph/chains/chain_data_service.py

@@ -1,8 +1,10 @@
 import asyncio
 from io import StringIO
-from typing import Dict, Optional, List, Any, Mapping, Set, cast, Type, Union
+from typing import Dict, Optional, List, Any, Mapping, Set, cast, Type, Union, Self
 
+import aio_pika.abc
 from aleph_message.models import StoreContent, ItemType, Chain, MessageType
+from configmanager import Config
 from pydantic import ValidationError
 
 from aleph.chains.common import LOGGER
@@ -36,7 +38,9 @@
 
 class ChainDataService:
     def __init__(
-        self, session_factory: DbSessionFactory, storage_service: StorageService
+        self,
+        session_factory: DbSessionFactory,
+        storage_service: StorageService,
     ):
         self.session_factory = session_factory
         self.storage_service = storage_service
@@ -215,11 +219,54 @@ async def get_tx_messages(
                 LOGGER.info("%s", error_msg)
                 raise InvalidContent(error_msg)
 
+
+async def make_pending_tx_exchange(config: Config) -> aio_pika.abc.AbstractExchange:
+    mq_conn = await aio_pika.connect_robust(
+        host=config.rabbitmq.host.value,
+        port=config.rabbitmq.port.value,
+        login=config.rabbitmq.username.value,
+        password=config.rabbitmq.password.value,
+    )
+    channel = await mq_conn.channel()
+    pending_tx_exchange = await channel.declare_exchange(
+        name=config.rabbitmq.pending_tx_exchange.value,
+        type=aio_pika.ExchangeType.TOPIC,
+        auto_delete=False,
+    )
+    return pending_tx_exchange
+
+
+class PendingTxPublisher:
+    def __init__(self, pending_tx_exchange: aio_pika.abc.AbstractExchange):
+        self.pending_tx_exchange = pending_tx_exchange
+
     @staticmethod
-    async def incoming_chaindata(session: DbSession, tx: ChainTxDb):
-        """Incoming data from a chain.
-        Content can be inline of "offchain" through an ipfs hash.
-        For now, we only add it to the database, it will be processed later.
-        """
+    def add_pending_tx(session: DbSession, tx: ChainTxDb):
         upsert_chain_tx(session=session, tx=tx)
         upsert_pending_tx(session=session, tx_hash=tx.hash)
+
+    async def publish_pending_tx(self, tx: ChainTxDb):
+        message = aio_pika.Message(body=tx.hash.encode("utf-8"))
+        await self.pending_tx_exchange.publish(
+            message=message, routing_key=f"{tx.chain.value}.{tx.publisher}.{tx.hash}"
+        )
+
+    async def add_and_publish_pending_tx(self, session: DbSession, tx: ChainTxDb):
+        """
+        Add an event published on one of the supported chains.
+        Adds the tx to the database, creates a pending tx entry in the pending tx table
+        and publishes a message on the pending tx exchange.
+
+        Note that this function commits changes to the database for consistency
+        between the DB and the message queue.
+        """
+        self.add_pending_tx(session=session, tx=tx)
+        session.commit()
+        await self.publish_pending_tx(tx)
+
+    @classmethod
+    async def new(cls, config: Config) -> Self:
+        pending_tx_exchange = await make_pending_tx_exchange(config=config)
+        return cls(
+            pending_tx_exchange=pending_tx_exchange,
+        )

src/aleph/chains/connector.py

@@ -5,11 +5,10 @@
 from aleph_message.models import Chain
 from configmanager import Config
 
-from aleph.storage import StorageService
 from aleph.types.db_session import DbSessionFactory
-from .bsc import BscConnector
-from .chain_data_service import ChainDataService
 from .abc import ChainReader, ChainWriter
+from .bsc import BscConnector
+from .chain_data_service import ChainDataService, PendingTxPublisher
 from .ethereum import EthereumConnector
 from .nuls2 import Nuls2Connector
 from .tezos import TezosConnector
@@ -28,9 +27,13 @@ class ChainConnector:
     writers: Dict[Chain, ChainWriter]
 
     def __init__(
-        self, session_factory: DbSessionFactory, chain_data_service: ChainDataService
+        self,
+        session_factory: DbSessionFactory,
+        pending_tx_publisher: PendingTxPublisher,
+        chain_data_service: ChainDataService,
     ):
         self._session_factory = session_factory
+        self.pending_tx_publisher = pending_tx_publisher
         self._chain_data_service = chain_data_service
 
         self.readers = {}
@@ -101,27 +104,29 @@ def _register_chains(self):
             Chain.BSC,
             BscConnector(
                 session_factory=self._session_factory,
-                chain_data_service=self._chain_data_service,
+                pending_tx_publisher=self.pending_tx_publisher,
             ),
         )
         self._add_chain(
             Chain.NULS2,
             Nuls2Connector(
                 session_factory=self._session_factory,
+                pending_tx_publisher=self.pending_tx_publisher,
                 chain_data_service=self._chain_data_service,
             ),
         )
         self._add_chain(
             Chain.ETH,
             EthereumConnector(
                 session_factory=self._session_factory,
+                pending_tx_publisher=self.pending_tx_publisher,
                 chain_data_service=self._chain_data_service,
             ),
         )
         self._add_chain(
             Chain.TEZOS,
             TezosConnector(
                 session_factory=self._session_factory,
-                chain_data_service=self._chain_data_service,
+                pending_tx_publisher=self.pending_tx_publisher,
             ),
         )

src/aleph/chains/ethereum.py

@@ -22,16 +22,16 @@
 from aleph.db.accessors.messages import get_unconfirmed_messages
 from aleph.db.accessors.pending_messages import count_pending_messages
 from aleph.db.accessors.pending_txs import count_pending_txs
+from aleph.db.models.chains import ChainTxDb
 from aleph.schemas.chains.tx_context import TxContext
 from aleph.schemas.pending_messages import BasePendingMessage
 from aleph.toolkit.timestamp import utc_now
+from aleph.types.chain_sync import ChainEventType
 from aleph.types.db_session import DbSessionFactory
 from aleph.utils import run_in_executor
-from .chain_data_service import ChainDataService
-from .abc import ChainWriter, Verifier, ChainReader
+from .abc import ChainWriter, Verifier
+from .chain_data_service import ChainDataService, PendingTxPublisher
 from .indexer_reader import AlephIndexerReader
-from ..db.models import ChainTxDb
-from ..types.chain_sync import ChainEventType
 
 LOGGER = logging.getLogger("chains.ethereum")
 CHAIN_NAME = "ETH"
@@ -105,15 +105,17 @@ class EthereumConnector(ChainWriter):
     def __init__(
         self,
         session_factory: DbSessionFactory,
+        pending_tx_publisher: PendingTxPublisher,
         chain_data_service: ChainDataService,
     ):
         self.session_factory = session_factory
+        self.pending_tx_publisher = pending_tx_publisher
         self.chain_data_service = chain_data_service
 
         self.indexer_reader = AlephIndexerReader(
             chain=Chain.ETH,
             session_factory=session_factory,
-            chain_data_service=chain_data_service,
+            pending_tx_publisher=pending_tx_publisher,
         )
 
     async def get_last_height(self, sync_type: ChainEventType) -> int:
@@ -212,7 +214,9 @@ async def _request_transactions(
 
                 except json.JSONDecodeError:
                     # if it's not valid json, just ignore it...
-                    LOGGER.info("Incoming logic data is not JSON, ignoring. %r" % message)
+                    LOGGER.info(
+                        "Incoming logic data is not JSON, ignoring. %r" % message
+                    )
 
                 except Exception:
                     LOGGER.exception("Can't decode incoming logic data %r" % message)
@@ -256,7 +260,7 @@ async def fetch_ethereum_sync_events(self, config: Config):
             ):
                 tx = ChainTxDb.from_sync_tx_context(tx_context=context, tx_data=jdata)
                 with self.session_factory() as session:
-                    await self.chain_data_service.incoming_chaindata(
+                    await self.pending_tx_publisher.add_and_publish_pending_tx(
                         session=session, tx=tx
                     )
                     session.commit()
@@ -313,7 +317,6 @@ async def packer(self, config: Config):
         gas_price = web3.eth.generate_gas_price()
         while True:
             with self.session_factory() as session:
-
                 # Wait for sync operations to complete
                 if (count_pending_txs(session=session, chain=Chain.ETH)) or (
                     count_pending_messages(session=session, chain=Chain.ETH)
@@ -344,8 +347,10 @@ async def packer(self, config: Config):
                 LOGGER.info("Chain sync: %d unconfirmed messages")
 
                 # This function prepares a chain data file and makes it downloadable from the node.
-                sync_event_payload = await self.chain_data_service.prepare_sync_event_payload(
-                    session=session, messages=messages
+                sync_event_payload = (
+                    await self.chain_data_service.prepare_sync_event_payload(
+                        session=session, messages=messages
+                    )
                 )
                 # Required to apply update to the files table in get_chaindata
                 session.commit()

src/aleph/chains/indexer_reader.py

@@ -21,7 +21,7 @@
 from pydantic import BaseModel
 
 import aleph.toolkit.json as aleph_json
-from aleph.chains.chain_data_service import ChainDataService
+from aleph.chains.chain_data_service import PendingTxPublisher
 from aleph.db.accessors.chains import (
     get_missing_indexer_datetime_multirange,
     add_indexer_range,
@@ -154,7 +154,6 @@ async def fetch_account_state(
         blockchain: IndexerBlockchain,
         accounts: List[str],
     ) -> IndexerAccountStateResponse:
-
         query = make_account_state_query(
             blockchain=blockchain, accounts=accounts, type_=EntityType.LOG
         )
@@ -194,7 +193,6 @@ def indexer_event_to_chain_tx(
     chain: Chain,
     indexer_event: Union[MessageEvent, SyncEvent],
 ) -> ChainTxDb:
-
     if isinstance(indexer_event, MessageEvent):
         protocol = ChainSyncProtocol.SMART_CONTRACT
         protocol_version = 1
@@ -225,7 +223,6 @@ async def extract_aleph_messages_from_indexer_response(
     chain: Chain,
     indexer_response: IndexerEventResponse,
 ) -> List[ChainTxDb]:
-
     message_events = indexer_response.data.message_events
     sync_events = indexer_response.data.sync_events
 
@@ -240,7 +237,6 @@ async def extract_aleph_messages_from_indexer_response(
 
 
 class AlephIndexerReader:
-
     BLOCKCHAIN_MAP: Mapping[Chain, IndexerBlockchain] = {
         Chain.BSC: IndexerBlockchain.BSC,
         Chain.ETH: IndexerBlockchain.ETHEREUM,
@@ -251,11 +247,11 @@ def __init__(
         self,
         chain: Chain,
         session_factory: DbSessionFactory,
-        chain_data_service: ChainDataService,
+        pending_tx_publisher: PendingTxPublisher,
     ):
         self.chain = chain
         self.session_factory = session_factory
-        self.chain_data_service = chain_data_service
+        self.pending_tx_publisher = pending_tx_publisher
 
         self.blockchain = self.BLOCKCHAIN_MAP[chain]
 
@@ -299,9 +295,7 @@ async def fetch_range(
                 LOGGER.info("%d new txs", len(txs))
                 # Events are listed in reverse order in the indexer response
                 for tx in txs:
-                    await self.chain_data_service.incoming_chaindata(
-                        session=session, tx=tx
-                    )
+                    self.pending_tx_publisher.add_pending_tx(session=session, tx=tx)
 
                 if nb_events_fetched >= limit:
                     last_event_datetime = txs[-1].datetime
@@ -317,6 +311,7 @@ async def fetch_range(
                 )
             else:
                 synced_range = Range(start_datetime, end_datetime, upper_inc=True)
+                txs = []
 
             LOGGER.info(
                 "%s %s indexer: fetched %s",
@@ -336,6 +331,10 @@ async def fetch_range(
             # of events.
             session.commit()
 
+            # Now that the txs are committed to the DB, add them to the pending tx message queue
+            for tx in txs:
+                await self.pending_tx_publisher.publish_pending_tx(tx)
+
             if nb_events_fetched < limit:
                 LOGGER.info(
                     "%s %s event indexer: done fetching events.",

src/aleph/chains/nuls2.py

@@ -30,7 +30,7 @@
 from aleph.toolkit.timestamp import utc_now
 from aleph.types.db_session import DbSessionFactory
 from aleph.utils import run_in_executor
-from .chain_data_service import ChainDataService
+from .chain_data_service import ChainDataService, PendingTxPublisher
 from .abc import Verifier, ChainWriter
 from aleph.schemas.chains.tx_context import TxContext
 from ..db.models import ChainTxDb
@@ -78,9 +78,13 @@ async def verify_signature(self, message: BasePendingMessage) -> bool:
 
 class Nuls2Connector(ChainWriter):
     def __init__(
-        self, session_factory: DbSessionFactory, chain_data_service: ChainDataService
+        self,
+        session_factory: DbSessionFactory,
+        pending_tx_publisher: PendingTxPublisher,
+        chain_data_service: ChainDataService,
     ):
         self.session_factory = session_factory
+        self.pending_tx_publisher = pending_tx_publisher
         self.chain_data_service = chain_data_service
 
     async def get_last_height(self, sync_type: ChainEventType) -> int:
@@ -154,7 +158,7 @@ async def fetcher(self, config: Config):
                         tx_context=context, tx_data=jdata
                     )
                     with self.session_factory() as db_session:
-                        await self.chain_data_service.incoming_chaindata(
+                        await self.pending_tx_publisher.add_and_publish_pending_tx(
                             session=db_session, tx=tx
                         )
                         db_session.commit()
@@ -197,8 +201,10 @@ async def packer(self, config: Config):
 
             if len(messages):
                 # This function prepares a chain data file and makes it downloadable from the node.
-                sync_event_payload = await self.chain_data_service.prepare_sync_event_payload(
-                    session=session, messages=messages
+                sync_event_payload = (
+                    await self.chain_data_service.prepare_sync_event_payload(
+                        session=session, messages=messages
+                    )
                 )
                 # Required to apply update to the files table in get_chaindata
                 session.commit()