Consume events from the topic exchange

AGENTS.md

@@ -99,8 +99,10 @@ Three processes share the same code:
 1. **Web** (`app/main.py`): FastAPI app, routers mounted under `/api/v1`, plus the unprefixed
    `default` router (docs, health) and the sqladmin panel at `/admin`.
 2. **Queue consumer**: started inside the FastAPI lifespan, not a separate process. `QueueProvider`
-   binds a durable queue to the `safe-transaction-service-events` fanout exchange and
-   `EventsService.process_event` turns each event into taskiq tasks.
+   binds a durable queue to the `safe-transaction-service-events-with-topics` topic exchange
+   with one `*.<TYPE>.*` binding per event type in `CONSUMED_EVENT_TYPES` (the types
+   `EventsService` handles), and `EventsService.process_event` turns each event into taskiq
+   tasks.
 3. **Taskiq worker + scheduler** (`app/workers/tasks.py`): `docker/web/taskiq/worker/run.sh` runs
    both, backed by Redis Streams.
 
@@ -221,8 +223,8 @@ Environment variables (see `.env.sample` and `app/config.py`):
   connections are reclaimed (default: 30000)
 - `REDIS_URL`: Redis connection string, used for the response cache and the taskiq broker
 - `RABBITMQ_AMQP_URL`: RabbitMQ connection for the event queue
-- `RABBITMQ_AMQP_EXCHANGE`: Fanout exchange published by Transaction Service
-  (default: `safe-transaction-service-events`)
+- `RABBITMQ_AMQP_EXCHANGE`: Topic exchange published by Transaction Service, declared as
+  durable `topic` (default: `safe-transaction-service-events-with-topics`)
 - `RABBITMQ_DECODER_EVENTS_QUEUE_NAME`: Queue name for this service (default: `safe-decoder-service`)
 - `ETHERSCAN_API_KEY`: API key for the Etherscan V2 client
 - `ETHERSCAN_MAX_REQUESTS` / `BLOCKSCOUT_MAX_REQUESTS` / `SOURCIFY_MAX_REQUESTS`: Concurrent

app/config.py

@@ -35,7 +35,7 @@ class Settings(BaseSettings):
     # Force-closes transactions left idle so leaked connections are reclaimed
     DATABASE_IDLE_IN_TRANSACTION_SESSION_TIMEOUT_MS: int = 30_000
     RABBITMQ_AMQP_URL: str = "amqp://guest:guest@"
-    RABBITMQ_AMQP_EXCHANGE: str = "safe-transaction-service-events"
+    RABBITMQ_AMQP_EXCHANGE: str = "safe-transaction-service-events-with-topics"
     RABBITMQ_DECODER_EVENTS_QUEUE_NAME: str = "safe-decoder-service"
     SECRET_KEY: str = secrets.token_urlsafe(
         32

app/datasources/queue/queue_provider.py

@@ -27,6 +27,11 @@
 # Name shown by RabbitMQ in the management UI and logs. The hostname is the pod name
 # in Kubernetes, so every replica can be told apart.
 AMQP_CONNECTION_NAME = f"safe-decoder-service@{socket.gethostname()}"
+# Event types the service consumes. `EventsService` processes only these types,
+# so only these are bound and the queue does not fill up with unused events
+CONSUMED_EVENT_TYPES = ("EXECUTED_MULTISIG_TRANSACTION",)
+# Routing keys are `{chainId}.{type}.{address}`
+EVENTS_BINDING_KEYS = tuple(f"*.{event_type}.*" for event_type in CONSUMED_EVENT_TYPES)
 
 
 class QueueProvider:
@@ -61,15 +66,17 @@ async def _connect(self, loop: AbstractEventLoop) -> None:
             raise QueueProviderUnableToConnectException from e
 
         channel = await self._connection.channel()
+        # Type and durability must match the exchange declared by Transaction
+        # Service, or RabbitMQ closes the channel with PRECONDITION_FAILED
         self._exchange = await channel.declare_exchange(
-            settings.RABBITMQ_AMQP_EXCHANGE, ExchangeType.FANOUT, durable=True
+            settings.RABBITMQ_AMQP_EXCHANGE, ExchangeType.TOPIC, durable=True
         )
         logger.info("Connected to %s exchange", settings.RABBITMQ_AMQP_EXCHANGE)
         self._events_queue = await channel.declare_queue(
             settings.RABBITMQ_DECODER_EVENTS_QUEUE_NAME, durable=True
         )
-        if self._events_queue:
-            await self._events_queue.bind(self._exchange)
+        for binding_key in EVENTS_BINDING_KEYS:
+            await self._events_queue.bind(self._exchange, routing_key=binding_key)
         logger.info(
             "Reading from %s queue", settings.RABBITMQ_DECODER_EVENTS_QUEUE_NAME
         )

app/services/events.py

@@ -8,6 +8,7 @@
 from safe_eth.eth.utils import fast_is_checksum_address
 from safe_eth.safe.multi_send import MultiSend
 
+from ..datasources.queue.queue_provider import CONSUMED_EVENT_TYPES
 from ..workers.tasks import (
     create_safe_contracts_task_for_new_chains,
     get_contract_metadata_task,
@@ -82,7 +83,7 @@ def _is_processable_event(tx_service_event: dict) -> bool:
 
         return (
             isinstance(event_type, str)
-            and event_type == "EXECUTED_MULTISIG_TRANSACTION"
+            and event_type in CONSUMED_EVENT_TYPES
             and isinstance(chain_id, str)
             and chain_id.isdigit()
             and (data is None or isinstance(data, str))

app/tests/datasources/queue/test_queue_provider.py

@@ -5,9 +5,7 @@
 from unittest.mock import patch
 
 import aio_pika
-from aio_pika.abc import AbstractRobustConnection
 
-from app.config import settings
 from app.datasources.queue.exceptions import QueueProviderUnableToConnectException
 from app.datasources.queue.queue_provider import AMQP_CONNECTION_NAME, QueueProvider
 
@@ -17,6 +15,9 @@ async def asyncSetUp(self):
         self.provider = QueueProvider()
         self.loop = asyncio.get_event_loop()
 
+    async def asyncTearDown(self):
+        await self.provider.disconnect()
+
     async def test_connect_success(self):
         self.assertFalse(self.provider.is_connected())
         await self.provider.connect(self.loop)
@@ -44,29 +45,40 @@ async def test_connect_failure(self):
             with self.assertRaises(QueueProviderUnableToConnectException):
                 await provider.connect(self.loop)
 
-    async def test_consume(self):
+    async def test_consume_only_bound_event_types(self):
         await self.provider.connect(self.loop)
-        assert isinstance(self.provider._connection, AbstractRobustConnection)
-        message = "Test message"
-        channel = await self.provider._connection.channel()
-        exchange = await channel.declare_exchange(
-            settings.RABBITMQ_AMQP_EXCHANGE, aio_pika.ExchangeType.FANOUT, durable=True
-        )
+        assert self.provider._exchange is not None
+        assert self.provider._events_queue is not None
+        await self.provider._events_queue.purge()
+        # The last message is bound. RabbitMQ keeps the publish order inside a
+        # queue, so once it is received any earlier message routed to the
+        # queue was received too
+        routing_keys = [
+            "1.INCOMING_ETHER.0x5afe",
+            "1.EXECUTED_MULTISIG_TRANSACTION.0x5afe",
+            "1.PENDING_MULTISIG_TRANSACTION.0x5afe",
+            "EXECUTED_MULTISIG_TRANSACTION",
+            "100.EXECUTED_MULTISIG_TRANSACTION._",
+        ]
+        expected_messages = [
+            "1.EXECUTED_MULTISIG_TRANSACTION.0x5afe",
+            "100.EXECUTED_MULTISIG_TRANSACTION._",
+        ]
+        for routing_key in routing_keys:
+            await self.provider._exchange.publish(
+                aio_pika.Message(body=routing_key.encode("utf-8")),
+                routing_key=routing_key,
+            )
 
-        await exchange.publish(
-            aio_pika.Message(body=message.encode("utf-8")),
-            routing_key="",
-        )
-
-        received_messages = []
+        received_messages: list[str] = []
+        last_received = asyncio.Event()
 
         async def callback(message: str):
             received_messages.append(message)
+            if message == expected_messages[-1]:
+                last_received.set()
 
         await self.provider.consume(callback)
+        await asyncio.wait_for(last_received.wait(), timeout=5)
 
-        # Wait to make sure the message is consumed.
-        await asyncio.sleep(1)
-
-        self.assertIn(message, received_messages)
-        await self.provider.disconnect()
+        self.assertEqual(received_messages, expected_messages)