diff --git a/packages/service-library/src/servicelib/rabbitmq/_client.py b/packages/service-library/src/servicelib/rabbitmq/_client.py index 25153e621f44..8c792e6efb87 100644 --- a/packages/service-library/src/servicelib/rabbitmq/_client.py +++ b/packages/service-library/src/servicelib/rabbitmq/_client.py @@ -1,16 +1,20 @@ import asyncio +import datetime import logging from dataclasses import dataclass, field from functools import partial -from typing import Annotated, Final +from typing import Annotated, Any, Final from uuid import uuid4 import aio_pika from aiormq import ChannelInvalidStateError +from aiormq.exceptions import ChannelNotFoundEntity from annotated_types import doc +from common_library.async_tools import cancel_wait_task from common_library.logging.logging_errors import create_troubleshooting_log_kwargs -from pydantic import NonNegativeInt +from pydantic import NonNegativeInt, PositiveInt +from ..background_task import create_periodic_task from ..logging_utils import log_catch, log_context from ._client_base import RabbitMQClientBase from ._models import ( @@ -34,6 +38,9 @@ _DEFAULT_RABBITMQ_EXECUTION_TIMEOUT_S: Final[int] = 5 _HEADER_X_DEATH: Final[str] = "x-death" +_BACKLOG_MONITOR_INTERVAL: Final[datetime.timedelta] = datetime.timedelta(seconds=60) +_BACKLOG_MONITOR_CONSECUTIVE_GROWTH_TO_WARN: Final[int] = 3 + _DEFAULT_UNEXPECTED_ERROR_RETRY_DELAY_S: Final[float] = 1 _DEFAULT_UNEXPECTED_ERROR_MAX_ATTEMPTS: Final[NonNegativeInt] = 15 @@ -56,7 +63,20 @@ async def _nack_message( message_handler: MessageHandler, max_retries_upon_error: int, message: aio_pika.abc.AbstractIncomingMessage, + *, + dead_letter_requeue_enabled: bool, ) -> None: + if not dead_letter_requeue_enabled: + # NOTE: no dead-letter-exchange configured on this queue: nacking here drops the message + # for good, there is no retry to report on + _logger.warning( + "Handler '%s' failed for message_id='%s'; dropping it (dead-letter requeue disabled)", + message_handler, + message.message_id, + ) + await message.nack(requeue=False) + return + count = _get_x_death_count(message) _logger.debug( "Nacking message '%s' from handler '%s', death count %s, max retries %s", @@ -88,6 +108,8 @@ async def _on_message( message_handler: MessageHandler, max_retries_upon_error: int, message: aio_pika.abc.AbstractIncomingMessage, + *, + dead_letter_requeue_enabled: bool, ) -> None: log_error_context = { "message_id": message.message_id, @@ -113,7 +135,12 @@ async def _on_message( logging.DEBUG, msg=f"Nack message {message.exchange=}, {message.routing_key=}", ): - await _nack_message(message_handler, max_retries_upon_error, message) + await _nack_message( + message_handler, + max_retries_upon_error, + message, + dead_letter_requeue_enabled=dead_letter_requeue_enabled, + ) except Exception as exc: # pylint: disable=broad-exception-caught _logger.exception( **create_troubleshooting_log_kwargs( @@ -125,7 +152,12 @@ async def _on_message( ) ) with log_catch(_logger, reraise=False): - await _nack_message(message_handler, max_retries_upon_error, message) + await _nack_message( + message_handler, + max_retries_upon_error, + message, + dead_letter_requeue_enabled=dead_letter_requeue_enabled, + ) except ChannelInvalidStateError as exc: # NOTE: this error can happen as can be seen in aio-pika code @@ -146,6 +178,7 @@ async def _on_message( class RabbitMQClient(RabbitMQClientBase): _connection_pool: aio_pika.pool.Pool | None = field(init=False, default=None) _channel_pool: aio_pika.pool.Pool | None = field(init=False, default=None) + _backlog_monitor_tasks: dict[QueueName, asyncio.Task] = field(init=False, default_factory=dict) def __post_init__(self) -> None: # recommendations are 1 connection per process @@ -175,6 +208,14 @@ async def close(self) -> None: logging.INFO, msg=f"{self.client_name} closing connection to RabbitMQ", ): + with log_catch(_logger, reraise=False): + await asyncio.gather( + *( + cancel_wait_task(task, max_delay=_DEFAULT_RABBITMQ_EXECUTION_TIMEOUT_S) + for task in self._backlog_monitor_tasks.values() + ) + ) + self._backlog_monitor_tasks.clear() assert self._channel_pool # nosec await self._channel_pool.close() assert self._connection_pool # nosec @@ -188,10 +229,61 @@ async def _get_channel(self) -> aio_pika.abc.AbstractChannel: channel.close_callbacks.add(self._channel_close_callback) return channel + def _start_backlog_monitor(self, queue_name: QueueName, exchange_name: ExchangeName, max_length: int) -> None: + """Warns if a bounded queue's ready-message count keeps growing, i.e. the + consumer cannot keep up with the publish rate (see `subscribe`'s `max_length`).""" + previous_message_count: int | None = None + consecutive_growth = 0 + + async def _check_backlog() -> None: + nonlocal previous_message_count, consecutive_growth + assert self._connection_pool # nosec + # NOTE: uses a dedicated, short-lived channel (not the shared `_channel_pool`) since + # that pool can hand out a channel that is concurrently in use to actively deliver + # an unacked message to a consumer, causing protocol interleaving on that channel + async with self._connection_pool.acquire() as connection: + channel = await connection.channel() + try: + declared = await channel.declare_queue( + queue_name, passive=True, timeout=_DEFAULT_RABBITMQ_EXECUTION_TIMEOUT_S + ) + message_count = declared.declaration_result.message_count + except ChannelNotFoundEntity: + # queue is gone (e.g. unsubscribe() raced this check): stop monitoring for good + _logger.debug("Queue '%s' no longer exists, stopping its backlog monitor", queue_name) + task = self._backlog_monitor_tasks.pop(queue_name, None) + if task is not None: + task.cancel() + return + finally: + await channel.close() + consecutive_growth = ( + consecutive_growth + 1 + if previous_message_count is not None and message_count and message_count > previous_message_count + else 0 + ) + if consecutive_growth >= _BACKLOG_MONITOR_CONSECUTIVE_GROWTH_TO_WARN: + _logger.warning( + "Queue '%s' (exchange '%s') backlog kept growing for %s consecutive checks " + "(now %s/%s ready messages): the consumer may not be keeping up with the publish rate", + queue_name, + exchange_name, + consecutive_growth, + message_count, + max_length, + ) + previous_message_count = message_count + + self._backlog_monitor_tasks[queue_name] = create_periodic_task( + _check_backlog, + interval=_BACKLOG_MONITOR_INTERVAL, + task_name=f"rabbitmq_backlog_monitor_{queue_name}", + ) + async def _create_consumer_tag(self, exchange_name) -> ConsumerTag: return ConsumerTag(f"{get_rabbitmq_client_unique_name(self.client_name)}_{exchange_name}_{uuid4()}") - async def subscribe( + async def subscribe( # noqa: PLR0913 # pylint: disable=too-many-arguments self, exchange_name: ExchangeName, message_handler: Annotated[ @@ -232,17 +324,57 @@ async def subscribe( NonNegativeInt, doc( "Also acts as a soft timeout: if `message_handler` does not finish processing " - "the message before this is reached, the message will be redelivered" + "the message before this is reached, the message will be redelivered (or, if " + "`enable_dead_letter_requeue=False`, simply dropped)" ), ] = RABBIT_QUEUE_MESSAGE_DEFAULT_TTL_MS, + max_length: Annotated[ + NonNegativeInt | None, + doc( + "Caps the queue depth. Once reached, the OLDEST ready messages are silently " + "dropped (`x-overflow: drop-head`) to make room for new ones, protecting the " + "broker from unbounded memory growth if `message_handler` ever falls behind " + "the publish rate. None (default) leaves the queue unbounded; only set this " + "for exchanges where losing old messages is preferable to broker instability. " + "NOTE: RabbitMQ dead-letters messages dropped this way too (reason `maxlen`), same " + "as nacked/expired ones; pair this with `enable_dead_letter_requeue=False`, or the " + "dropped messages will bounce forever between this queue and its delay queue until " + "RabbitMQ's own dead-letter-cycle detector catches it" + ), + ] = None, + prefetch_count: Annotated[ + PositiveInt | None, + doc( + "Maximum number of messages delivered to (and awaiting ack from) `message_handler` " + "concurrently. None (default) uses 1 for shared (`exclusive_queue=False`) queues, or " + f"{_DEFAULT_PREFETCH_VALUE} for exclusive ones. Raise this for lightweight, I/O-bound " + "handlers on high-throughput exchanges, where the default is otherwise the throughput " + "ceiling regardless of how fast `message_handler` actually runs" + ), + ] = None, unexpected_error_retry_delay_s: Annotated[ float, - doc("Time to wait between each retry when `message_handler` raised or returned `False`"), + doc( + "Time to wait between each retry when `message_handler` raised or returned `False`. " + "Has no effect when `enable_dead_letter_requeue=False`: such messages are dropped " + "immediately instead of being retried" + ), ] = _DEFAULT_UNEXPECTED_ERROR_RETRY_DELAY_S, unexpected_error_max_attempts: Annotated[ int, doc("Maximum amount of retries when `message_handler` raised or returned `False`"), ] = _DEFAULT_UNEXPECTED_ERROR_MAX_ATTEMPTS, + enable_dead_letter_requeue: Annotated[ + bool, + doc( + "When True (default), messages that are nacked or that expire after sitting " + "`message_ttl` in the queue are bounced through a delay queue and re-published into " + "THIS SAME exchange for a retry, up to `unexpected_error_max_attempts` times. Set False " + "for best-effort/fire-and-forget exchanges (e.g. live UI notifications) where a stale " + "message has no value: expired/nacked messages are then simply dropped instead of " + "generating more publish traffic back into an exchange that may already be backlogged" + ), + ] = True, ) -> Annotated[ tuple[QueueName, ConsumerTag], doc("Returns the queue name and consumer tag of the subscription"), @@ -256,7 +388,11 @@ async def subscribe( assert self._channel_pool # nosec async with self._channel_pool.acquire() as channel: - qos_value = 1 if exclusive_queue is False else _DEFAULT_PREFETCH_VALUE + qos_value = ( + prefetch_count + if prefetch_count is not None + else (1 if exclusive_queue is False else _DEFAULT_PREFETCH_VALUE) + ) await channel.set_qos(qos_value) exchange = await channel.declare_exchange( @@ -271,41 +407,55 @@ async def subscribe( # exclusive means that the queue is only available for THIS very client # and will be deleted when the client disconnects # NOTE what is a dead letter exchange, see https://www.rabbitmq.com/dlx.html - delayed_exchange_name = _DELAYED_EXCHANGE_NAME.format(exchange_name=exchange_name) + queue_arguments: dict[str, Any] = {} + if enable_dead_letter_requeue: + delayed_exchange_name = _DELAYED_EXCHANGE_NAME.format(exchange_name=exchange_name) + queue_arguments["x-dead-letter-exchange"] = delayed_exchange_name + if max_length is not None: + queue_arguments["x-max-length"] = max_length + queue_arguments["x-overflow"] = "drop-head" queue = await declare_queue( channel, self.client_name, non_exclusive_queue_name or exchange_name, exclusive_queue=exclusive_queue, message_ttl=message_ttl, - arguments={"x-dead-letter-exchange": delayed_exchange_name}, + arguments=queue_arguments, ) if topics is None: await queue.bind(exchange, routing_key="") else: await asyncio.gather(*(queue.bind(exchange, routing_key=topic) for topic in topics)) - delayed_exchange = await channel.declare_exchange( - delayed_exchange_name, aio_pika.ExchangeType.FANOUT, durable=True - ) - delayed_queue_name = _DELAYED_QUEUE_NAME.format(queue_name=non_exclusive_queue_name or exchange_name) - - delayed_queue = await declare_queue( - channel, - self.client_name, - delayed_queue_name, - exclusive_queue=exclusive_queue, - message_ttl=int(unexpected_error_retry_delay_s * 1000), - arguments={"x-dead-letter-exchange": exchange.name}, - ) - await delayed_queue.bind(delayed_exchange) + if enable_dead_letter_requeue: + delayed_exchange = await channel.declare_exchange( + delayed_exchange_name, aio_pika.ExchangeType.FANOUT, durable=True + ) + delayed_queue_name = _DELAYED_QUEUE_NAME.format(queue_name=non_exclusive_queue_name or exchange_name) + + delayed_queue = await declare_queue( + channel, + self.client_name, + delayed_queue_name, + exclusive_queue=exclusive_queue, + message_ttl=int(unexpected_error_retry_delay_s * 1000), + arguments={"x-dead-letter-exchange": exchange.name}, + ) + await delayed_queue.bind(delayed_exchange) consumer_tag = await self._create_consumer_tag(exchange_name) await queue.consume( - partial(_on_message, message_handler, unexpected_error_max_attempts), + partial( + _on_message, + message_handler, + unexpected_error_max_attempts, + dead_letter_requeue_enabled=enable_dead_letter_requeue, + ), exclusive=exclusive_queue, consumer_tag=consumer_tag, ) + if max_length is not None: + self._start_backlog_monitor(queue.name, exchange_name, max_length) return queue.name, consumer_tag async def add_topics( @@ -323,7 +473,7 @@ async def add_topics( self.client_name, exchange_name, exclusive_queue=True, - arguments={"x-dead-letter-exchange": _DELAYED_EXCHANGE_NAME.format(exchange_name=exchange_name)}, + passive=True, ) await asyncio.gather(*(queue.bind(exchange, routing_key=topic) for topic in topics)) @@ -342,7 +492,7 @@ async def remove_topics( self.client_name, exchange_name, exclusive_queue=True, - arguments={"x-dead-letter-exchange": _DELAYED_EXCHANGE_NAME.format(exchange_name=exchange_name)}, + passive=True, ) await asyncio.gather( @@ -363,6 +513,9 @@ async def unsubscribe( queue = await channel.get_queue(queue_name) # NOTE: we force delete here await queue.delete(if_unused=False, if_empty=False) + backlog_monitor_task = self._backlog_monitor_tasks.pop(queue_name, None) + if backlog_monitor_task is not None: + await cancel_wait_task(backlog_monitor_task, max_delay=_DEFAULT_RABBITMQ_EXECUTION_TIMEOUT_S) async def publish(self, exchange_name: ExchangeName, message: RabbitMessage) -> None: """publish message in the exchange exchange_name. diff --git a/packages/service-library/src/servicelib/rabbitmq/_utils.py b/packages/service-library/src/servicelib/rabbitmq/_utils.py index c647f5f5ce2b..c84b25d99cbe 100644 --- a/packages/service-library/src/servicelib/rabbitmq/_utils.py +++ b/packages/service-library/src/servicelib/rabbitmq/_utils.py @@ -74,7 +74,20 @@ async def declare_queue( exclusive_queue: bool, arguments: dict[str, Any] | None = None, message_ttl: NonNegativeInt = RABBIT_QUEUE_MESSAGE_DEFAULT_TTL_MS, + passive: bool = False, ) -> aio_pika.abc.AbstractRobustQueue: + """Declares (or, if `passive=True`, just looks up) the queue derived from `queue_name`. + + `passive=False` (default): asks the broker to create the queue if it does not exist, or + otherwise assert that it exists AND has the exact same `durable`/`exclusive`/`arguments` as + passed here; a mismatch raises `ChannelPreconditionFailed` and closes the channel. Use this + only where this call is meant to define the queue's shape (e.g. the original `subscribe()`). + + `passive=True`: only checks that the queue already exists (raises if not) and returns a handle + to it; the broker does not create anything and does not compare arguments, so this is safe to + use to just re-attach to (e.g. to bind/unbind topics on) a queue defined elsewhere, even if the + `arguments` passed here don't match what that queue was actually created with. + """ default_arguments = {"x-message-ttl": message_ttl} if arguments is not None: default_arguments.update(arguments) @@ -82,6 +95,7 @@ async def declare_queue( "durable": not exclusive_queue, "exclusive": exclusive_queue, "arguments": default_arguments, + "passive": passive, "name": f"{get_rabbitmq_client_unique_name(client_name)}_{queue_name}_exclusive", } if not exclusive_queue: diff --git a/packages/service-library/tests/rabbitmq/test_rabbitmq.py b/packages/service-library/tests/rabbitmq/test_rabbitmq.py index 080d287b066c..9edd10a85f2f 100644 --- a/packages/service-library/tests/rabbitmq/test_rabbitmq.py +++ b/packages/service-library/tests/rabbitmq/test_rabbitmq.py @@ -6,6 +6,8 @@ import asyncio +import datetime +import logging from collections.abc import AsyncIterator, Awaitable, Callable from dataclasses import dataclass from typing import Any, Final @@ -703,3 +705,259 @@ async def test_unsubscribe_consumer( # Unsubscribe the queue for _ in range(idempotent_attempts): await client.unsubscribe(queue_name) + + +async def _wait_until_stable( + get_value: Callable[[], int], *, stable_polls: int = 3, poll_interval_s: float = 0.1, timeout_s: float = 5 +) -> int: + """Polls `get_value()` until it returns the same value `stable_polls` times in a row, + then returns that value. Deterministic replacement for a fixed sleep when waiting for + an asynchronous background process (e.g. draining a queue) to become quiescent.""" + deadline = asyncio.get_running_loop().time() + timeout_s + last_value = get_value() + stable_count = 1 + while stable_count < stable_polls: + if asyncio.get_running_loop().time() > deadline: + msg = f"value did not stabilize within {timeout_s}s (last seen: {last_value})" + raise TimeoutError(msg) + await asyncio.sleep(poll_interval_s) + current_value = get_value() + if current_value == last_value: + stable_count += 1 + else: + last_value = current_value + stable_count = 1 + return last_value + + +async def _wait_until_queue_drained( + connection_pool: aio_pika.pool.Pool, queue_name: QueueName, *, timeout_s: float = 5 +) -> None: + """Polls a queue's ready-message count via passive declare until it reaches zero, using a + dedicated channel (not the client's shared channel pool, to avoid interleaving with active + consumer traffic). Deterministic replacement for a fixed sleep before closing a connection. + + NOTE: not suitable for queues that ever hit `x-max-length` overflow (drop-head): RabbitMQ's + reported `message_count` can remain permanently off-by-N after such truncation even though the + queue is physically empty (confirmed independently via `basic.get`); for those, poll an + application-level counter (e.g. the handler's own received-count) instead. + """ + async with connection_pool.acquire() as connection: + channel = await connection.channel() + try: + async for attempt in AsyncRetrying( + wait=wait_fixed(0.05), + stop=stop_after_delay(timeout_s), + retry=retry_if_exception_type(AssertionError), + reraise=True, + ): + with attempt: + declared = await channel.declare_queue(queue_name, passive=True) + assert declared.declaration_result.message_count == 0 + finally: + await channel.close() + + +async def test_subscribe_with_max_length_drops_oldest_ready_messages( + create_rabbitmq_client: Callable[[str], RabbitMQClient], + random_exchange_name: Callable[[], str], + random_rabbit_message: Callable[..., PytestRabbitMessage], +): + consumer = create_rabbitmq_client("consumer") + publisher = create_rabbitmq_client("publisher") + exchange_name = random_exchange_name() + + block_processing = asyncio.Event() + received: list[bytes] = [] + + async def _blocking_handler(data: bytes) -> bool: + received.append(data) + await block_processing.wait() + return True + + max_length = 5 + queue_name, _ = await consumer.subscribe( + exchange_name, + _blocking_handler, + prefetch_count=1, + max_length=max_length, + # NOTE: RabbitMQ dead-letters messages dropped by `x-max-length` overflow too (reason + # "maxlen"), not just nacked/expired ones. Without this, dropped messages bounce forever + # between this queue and its delay queue until RabbitMQ's own dead-letter-cycle detector + # catches it - exactly why the two are always paired in production (see subscribe()'s docs) + enable_dead_letter_requeue=False, + ) + + num_messages = max_length + 10 + messages = [random_rabbit_message() for _ in range(num_messages)] + await asyncio.gather(*(publisher.publish(exchange_name, m) for m in messages)) + + # the first message is delivered and blocks the (single-prefetch) consumer; + # the rest pile up as ready messages, capped by max_length + async for attempt in AsyncRetrying( + wait=wait_fixed(0.1), + stop=stop_after_delay(5), + retry=retry_if_exception_type(AssertionError), + reraise=True, + ): + with attempt: + assert len(received) == 1 + + assert consumer._connection_pool # noqa: SLF001 + # NOTE: uses a dedicated channel, not consumer._channel_pool, since that shared pool + # could hand out the channel that is currently mid-delivery of the unacked message + async with consumer._connection_pool.acquire() as connection: # noqa: SLF001 + channel = await connection.channel() + declared = await channel.declare_queue(queue_name, passive=True) + assert declared.declaration_result.message_count is not None + assert declared.declaration_result.message_count <= max_length + await channel.close() + + block_processing.set() + # NOTE: deliberately NOT using `_wait_until_queue_drained` here: RabbitMQ's reported + # `message_count` can get permanently stuck above zero after `x-max-length`/drop-head + # truncation, even though the queue is physically empty. Polling the handler's own + # received-count is the reliable, deterministic ground truth for "processing finished". + async for attempt in AsyncRetrying( + wait=wait_fixed(0.05), + stop=stop_after_delay(5), + retry=retry_if_exception_type(AssertionError), + reraise=True, + ): + with attempt: + assert len(received) == max_length + 1 + # only the unacked message plus at most max_length ready ones ever got delivered + assert 1 <= len(received) <= max_length + 1 + assert len(received) < num_messages + + +async def test_subscribe_prefetch_count_limits_concurrent_deliveries( + create_rabbitmq_client: Callable[[str], RabbitMQClient], + random_exchange_name: Callable[[], str], + random_rabbit_message: Callable[..., PytestRabbitMessage], +): + consumer = create_rabbitmq_client("consumer") + publisher = create_rabbitmq_client("publisher") + exchange_name = random_exchange_name() + + release_processing = asyncio.Event() + in_flight = 0 + max_in_flight = 0 + + async def _handler(_: bytes) -> bool: + nonlocal in_flight, max_in_flight + in_flight += 1 + max_in_flight = max(max_in_flight, in_flight) + await release_processing.wait() + in_flight -= 1 + return True + + prefetch_count = 4 + queue_name, _consumer_tag = await consumer.subscribe(exchange_name, _handler, prefetch_count=prefetch_count) + + messages = [random_rabbit_message() for _ in range(prefetch_count * 3)] + await asyncio.gather(*(publisher.publish(exchange_name, m) for m in messages)) + + async for attempt in AsyncRetrying( + wait=wait_fixed(0.1), + stop=stop_after_delay(5), + retry=retry_if_exception_type(AssertionError), + reraise=True, + ): + with attempt: + assert max_in_flight == prefetch_count + + # give it a bit more time to ensure it never exceeds the configured prefetch + await asyncio.sleep(0.5) + assert max_in_flight == prefetch_count + + release_processing.set() + # deterministically wait for the queue to drain (all messages acked) before closing + assert consumer._connection_pool # noqa: SLF001 + await _wait_until_queue_drained(consumer._connection_pool, queue_name) # noqa: SLF001 + + +async def test_subscribe_enable_dead_letter_requeue_false_drops_failed_messages( + on_message_spy: mock.Mock, + create_rabbitmq_client: Callable[[str], RabbitMQClient], + random_exchange_name: Callable[[], str], + random_rabbit_message: Callable[..., PytestRabbitMessage], +): + publisher = create_rabbitmq_client("publisher") + consumer = create_rabbitmq_client("consumer") + exchange_name = random_exchange_name() + + async def _always_fail(_: Any) -> bool: + return False + + await consumer.subscribe( + exchange_name, + _always_fail, + enable_dead_letter_requeue=False, + unexpected_error_retry_delay_s=_ON_ERROR_DELAY_S, + ) + message = random_rabbit_message() + await publisher.publish(exchange_name, message) + + # with the retry machinery disabled, a failing message is delivered exactly once, never retried + await _assert_wait_for_messages(on_message_spy, 1) + + +async def test_subscribe_backlog_monitor_warns_when_consumer_falls_behind( + caplog: pytest.LogCaptureFixture, + mocker: MockerFixture, + create_rabbitmq_client: Callable[[str], RabbitMQClient], + random_exchange_name: Callable[[], str], + random_rabbit_message: Callable[..., PytestRabbitMessage], +): + mocker.patch.object(_client, "_BACKLOG_MONITOR_INTERVAL", datetime.timedelta(milliseconds=50)) + caplog.set_level(logging.WARNING) + + consumer = create_rabbitmq_client("consumer") + publisher = create_rabbitmq_client("publisher") + exchange_name = random_exchange_name() + + block_processing = asyncio.Event() + processed_count = 0 + + async def _blocking_handler(_: bytes) -> bool: + nonlocal processed_count + await block_processing.wait() + processed_count += 1 + return True + + _queue_name, _consumer_tag = await consumer.subscribe( + exchange_name, + _blocking_handler, + prefetch_count=1, + max_length=10000, + ) + + # publish continuously, much faster than every backlog check, so that the (blocked) + # consumer never drains anything and the ready count grows on every consecutive check + stop_publishing = asyncio.Event() + + async def _publish_forever() -> None: + while not stop_publishing.is_set(): + await publisher.publish(exchange_name, random_rabbit_message()) + + publish_task = asyncio.create_task(_publish_forever()) + try: + async for attempt in AsyncRetrying( + wait=wait_fixed(0.1), + stop=stop_after_delay(5), + retry=retry_if_exception_type(AssertionError), + reraise=True, + ): + with attempt: + assert "backlog kept growing" in caplog.text + finally: + stop_publishing.set() + await publish_task + block_processing.set() + # NOTE: deliberately NOT using `_wait_until_queue_drained` here: this queue's + # `max_length` overflow (drop-head) can leave RabbitMQ's reported `message_count` + # permanently stuck above zero even once the queue is physically empty. Polling the + # handler's own processed-count until it stops growing is the reliable, deterministic + # way to know the consumer has caught up before closing the connection. + await _wait_until_stable(lambda: processed_count, timeout_s=10) diff --git a/services/web/server/src/simcore_service_webserver/notifications/_rabbitmq_exclusive_queue_consumers.py b/services/web/server/src/simcore_service_webserver/notifications/_rabbitmq_exclusive_queue_consumers.py index 7b9ccb31ffd2..e36515badd2f 100644 --- a/services/web/server/src/simcore_service_webserver/notifications/_rabbitmq_exclusive_queue_consumers.py +++ b/services/web/server/src/simcore_service_webserver/notifications/_rabbitmq_exclusive_queue_consumers.py @@ -52,6 +52,20 @@ ) WALLET_SUBSCRIPTION_LOCK_APPKEY: Final = web.AppKey("WALLET_SUBSCRIPTION_LOCK", asyncio.Lock) +# NOTE: logs are high-volume and merely a UX nicety (unlike progress/pipeline-status/wallets +# events); if a replica's consumer ever falls behind (e.g. stale subscriptions piling up), +# cap the queue so dropping old log lines protects the broker instead of exhausting its memory. +# Chosen well above real, self-recovering broker-wide bursts observed in production over the +# past 45 days (up to ~920k ready messages within a single hour, always draining back down +# within ~1h) so normal spiky traffic is never clipped - this is a last-resort circuit breaker +# for genuine runaway growth (the 2026-08-26 incident reached ~7.4M before the broker crashed), +# not a routine control. +_LOGS_QUEUE_MAX_LENGTH: Final[int] = 1_500_000 +# NOTE: `_log_message_parser` is a cheap, I/O-bound handler (socket.io emit); the shared default +# of 10 in-flight messages is otherwise the hard throughput ceiling for this high-volume queue, +# regardless of how fast the handler itself runs +_LOGS_QUEUE_PREFETCH_COUNT: Final[int] = 100 + async def _notify_comp_node_progress(app: web.Application, message: ProgressRabbitMessageNode) -> None: project = await _projects_service.get_project_for_user( @@ -206,7 +220,14 @@ async def _osparc_credits_message_parser(app: web.Application, data: bytes) -> b SubscribeArgumentsTuple( LoggerRabbitMessage.get_channel_name(), _log_message_parser, - {"topics": []}, + { + "topics": [], + "max_length": _LOGS_QUEUE_MAX_LENGTH, + "prefetch_count": _LOGS_QUEUE_PREFETCH_COUNT, + # a stale log line has no value and dead-lettering it back into this same exchange + # would only add more publish traffic to an already-backlogged queue + "enable_dead_letter_requeue": False, + }, ), SubscribeArgumentsTuple( ProgressRabbitMessageNode.get_channel_name(), diff --git a/services/web/server/tests/unit/with_dbs/02/test_projects_nodes_handler.py b/services/web/server/tests/unit/with_dbs/02/test_projects_nodes_handler.py index 9eee0f26eff5..1ca003997595 100644 --- a/services/web/server/tests/unit/with_dbs/02/test_projects_nodes_handler.py +++ b/services/web/server/tests/unit/with_dbs/02/test_projects_nodes_handler.py @@ -355,7 +355,6 @@ def standard_user_role() -> tuple[str, tuple]: return (all_roles[0], (pytest.param(*all_roles[1][2], id="standard user role"),)) -@pytest.mark.flaky(max_runs=3) @pytest.mark.parametrize(*standard_user_role()) async def test_create_and_delete_many_nodes_in_parallel( mock_dynamic_scheduler: None,