Fix RabbitMQ connection timeouts during long operations. (#856)

Author: aliel

src/aleph/chains/chain_data_service.py

@@ -231,6 +231,7 @@ async def make_pending_tx_exchange(config: Config) -> aio_pika.abc.AbstractExcha
         port=config.rabbitmq.port.value,
         login=config.rabbitmq.username.value,
         password=config.rabbitmq.password.value,
+        heartbeat=config.rabbitmq.heartbeat.value,
     )
     channel = await mq_conn.channel()
     pending_tx_exchange = await channel.declare_exchange(

src/aleph/config.py

@@ -204,6 +204,8 @@ def get_defaults():
             "pending_message_exchange": "aleph-pending-messages",
             # Name of the RabbitMQ exchange used for sync/message events (input of the TX processor).
             "pending_tx_exchange": "aleph-pending-txs",
+            # Heartbeat interval in seconds to prevent connection timeouts during long operations.
+            "heartbeat": 600,
         },
         "redis": {
             # Hostname of the Redis service.

src/aleph/jobs/job_utils.py

@@ -41,6 +41,7 @@ async def _make_pending_queue(
             port=config.rabbitmq.port.value,
             login=config.rabbitmq.username.value,
             password=config.rabbitmq.password.value,
+            heartbeat=config.rabbitmq.heartbeat.value,
         )
         channel = await mq_conn.channel()
 

src/aleph/jobs/process_pending_messages.py

@@ -63,9 +63,14 @@ async def new(
         mq_password: str,
         message_exchange_name: str,
         pending_message_exchange_name: str,
+        mq_heartbeat: int,
     ):
         mq_conn = await aio_pika.connect_robust(
-            host=mq_host, port=mq_port, login=mq_username, password=mq_password
+            host=mq_host,
+            port=mq_port,
+            login=mq_username,
+            password=mq_password,
+            heartbeat=mq_heartbeat,
         )
         channel = await mq_conn.channel()
         mq_message_exchange = await channel.declare_exchange(
@@ -179,6 +184,7 @@ async def fetch_and_process_messages_task(config: Config):
             mq_password=config.rabbitmq.password.value,
             message_exchange_name=config.rabbitmq.message_exchange.value,
             pending_message_exchange_name=config.rabbitmq.pending_message_exchange.value,
+            mq_heartbeat=config.rabbitmq.heartbeat.value,
         )
 
         async with pending_message_processor:

src/aleph/toolkit/rabbitmq.py

@@ -7,5 +7,6 @@ async def make_mq_conn(config) -> aio_pika.abc.AbstractConnection:
         port=config.rabbitmq.port.value,
         login=config.rabbitmq.username.value,
         password=config.rabbitmq.password.value,
+        heartbeat=config.rabbitmq.heartbeat.value,
     )
     return mq_conn