From 192b37c0c67e0f5247343c0b12045ec71c67557e Mon Sep 17 00:00:00 2001 From: Uxio Fuentefria Date: Tue, 6 Oct 2026 14:31:30 +0000 Subject: [PATCH] Consume events from the topic exchange Bind the queue to the Transaction Service topic exchange only for EXECUTED_MULTISIG_TRANSACTION, the one event type the service processes. --- AGENTS.md | 10 ++-- app/config.py | 2 +- app/datasources/queue/queue_provider.py | 13 +++-- app/services/events.py | 3 +- .../datasources/queue/test_queue_provider.py | 52 ++++++++++++------- 5 files changed, 51 insertions(+), 29 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 9270e2c..dfd8562 100644 --- a/AGENTS.md +++ b/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 `*..*` 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 diff --git a/app/config.py b/app/config.py index dd5d01d..ffd7f3c 100644 --- a/app/config.py +++ b/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 diff --git a/app/datasources/queue/queue_provider.py b/app/datasources/queue/queue_provider.py index 3615f96..2ebe9c8 100644 --- a/app/datasources/queue/queue_provider.py +++ b/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 ) diff --git a/app/services/events.py b/app/services/events.py index 48e94ce..4d6633a 100644 --- a/app/services/events.py +++ b/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)) diff --git a/app/tests/datasources/queue/test_queue_provider.py b/app/tests/datasources/queue/test_queue_provider.py index 5e45a78..3506243 100644 --- a/app/tests/datasources/queue/test_queue_provider.py +++ b/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)