wip: made queues durable

Author: Andrei Neagu

packages/service-library/src/servicelib/deferred_tasks/_deferred_manager.py

@@ -8,7 +8,13 @@
 
 import arrow
 from faststream.exceptions import NackMessage, RejectMessage
-from faststream.rabbit import ExchangeType, RabbitBroker, RabbitExchange, RabbitRouter
+from faststream.rabbit import (
+    ExchangeType,
+    RabbitBroker,
+    RabbitExchange,
+    RabbitQueue,
+    RabbitRouter,
+)
 from pydantic import NonNegativeInt
 from servicelib.logging_utils import log_catch, log_context
 from servicelib.redis import RedisClientSDK
@@ -149,10 +155,14 @@ def __init__(
         self._global_resources_prefix = f"{calling_module_name}"
 
         self.common_exchange = RabbitExchange(
-            f"{self._global_resources_prefix}_common", type=ExchangeType.DIRECT
+            f"{self._global_resources_prefix}_common",
+            durable=True,
+            type=ExchangeType.DIRECT,
         )
         self.cancellation_exchange = RabbitExchange(
-            f"{self._global_resources_prefix}_cancellation", type=ExchangeType.FANOUT
+            f"{self._global_resources_prefix}_cancellation",
+            durable=True,
+            type=ExchangeType.FANOUT,
         )
 
     def patch_based_deferred_handlers(self) -> None:
@@ -243,8 +253,10 @@ def un_patch_base_deferred_handlers(cls) -> None:
                         subclass.is_present.original_is_present  # type: ignore
                     )
 
-    def _get_global_queue_name(self, queue_name: _FastStreamRabbitQueue) -> str:
-        return f"{self._global_resources_prefix}_{queue_name}"
+    def _get_global_queue(self, queue_name: _FastStreamRabbitQueue) -> RabbitQueue:
+        return RabbitQueue(
+            f"{self._global_resources_prefix}_{queue_name}", durable=True
+        )
 
     def __get_subclass(
         self, class_unique_reference: ClassUniqueReference
@@ -259,7 +271,7 @@ async def __publish_to_queue(
     ) -> None:
         await self.broker.publish(
             task_uid,
-            queue=self._get_global_queue_name(queue),
+            queue=self._get_global_queue(queue),
             exchange=(
                 self.cancellation_exchange
                 if queue == _FastStreamRabbitQueue.MANUALLY_CANCELLED
@@ -569,47 +581,43 @@ def _register_subscribers(self) -> None:
         # pylint:disable=unexpected-keyword-arg
         # pylint:disable=no-value-for-parameter
         self._fs_handle_scheduled = self.router.subscriber(
-            queue=self._get_global_queue_name(_FastStreamRabbitQueue.SCHEDULED),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.SCHEDULED),
             exchange=self.common_exchange,
             retry=True,
         )(self._fs_handle_scheduled)
 
         self._fs_handle_submit_task = self.router.subscriber(
-            queue=self._get_global_queue_name(_FastStreamRabbitQueue.SUBMIT_TASK),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.SUBMIT_TASK),
             exchange=self.common_exchange,
             retry=True,
         )(self._fs_handle_submit_task)
 
         self._fs_handle_worker = self.router.subscriber(
-            queue=self._get_global_queue_name(_FastStreamRabbitQueue.WORKER),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.WORKER),
             exchange=self.common_exchange,
             retry=True,
         )(self._fs_handle_worker)
 
         self._fs_handle_error_result = self.router.subscriber(
-            queue=self._get_global_queue_name(_FastStreamRabbitQueue.ERROR_RESULT),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.ERROR_RESULT),
             exchange=self.common_exchange,
             retry=True,
         )(self._fs_handle_error_result)
 
         self._fs_handle_finished_with_error = self.router.subscriber(
-            queue=self._get_global_queue_name(
-                _FastStreamRabbitQueue.FINISHED_WITH_ERROR
-            ),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.FINISHED_WITH_ERROR),
             exchange=self.common_exchange,
             retry=True,
         )(self._fs_handle_finished_with_error)
 
         self._fs_handle_deferred_result = self.router.subscriber(
-            queue=self._get_global_queue_name(_FastStreamRabbitQueue.DEFERRED_RESULT),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.DEFERRED_RESULT),
             exchange=self.common_exchange,
             retry=True,
         )(self._fs_handle_deferred_result)
 
         self._fs_handle_manually_cancelled = self.router.subscriber(
-            queue=self._get_global_queue_name(
-                _FastStreamRabbitQueue.MANUALLY_CANCELLED
-            ),
+            queue=self._get_global_queue(_FastStreamRabbitQueue.MANUALLY_CANCELLED),
             exchange=self.cancellation_exchange,
             retry=True,
         )(self._fs_handle_manually_cancelled)

packages/service-library/src/servicelib/rabbitmq/_client_rpc.py

@@ -49,7 +49,7 @@ async def _rpc_initialize(self) -> None:
         self._channel = await self._connection.channel()
 
         self._rpc = aio_pika.patterns.RPC(self._channel)
-        await self._rpc.initialize()
+        await self._rpc.initialize(durable=True)
 
     async def close(self) -> None:
         with log_context(
@@ -134,6 +134,7 @@ async def register_handler(
             RPCNamespacedMethodName.from_namespace_and_method(namespace, method_name),
             handler,
             auto_delete=True,
+            durable=True,
         )
 
     async def register_router(

packages/service-library/src/servicelib/rabbitmq/_utils.py

@@ -85,6 +85,23 @@ async def declare_queue(
         # NOTE: setting a name will ensure multiple instance will take their data here
         queue_parameters |= {"name": queue_name}
 
+    # avoids deprecated `transient_nonexcl_queues` warning
+    # TODO: raise an error here to avoid this configuration
+    if (
+        queue_parameters.get("durable", False) is False
+        and queue_parameters.get("exclusive", False) is False
+    ):
+        _logger.warning(
+            "Avoid `transient_nonexcl_queues` warning for queue_setup: %s",
+            queue_parameters,
+        )
+        # this is the closes equivalen of the old `transient_nonexcl_queues`
+        # without violating the deprecation warning
+        # https://github.com/rabbitmq/rabbitmq-server/discussions/13161
+        queue_parameters["durable"] = True
+        queue_parameters["message-ttl"] = 86400000  # 24 hours
+        _logger.warning("NEW params: %s", queue_parameters)
+
     # NOTE: if below line raises something similar to ``ChannelPreconditionFailed: PRECONDITION_FAILED``
     # most likely someone changed the signature of the queues (parameters etc...)
     # Safest way to deal with it:

packages/service-library/tests/deferred_tasks/test__utils.py

@@ -57,7 +57,7 @@ async def _() -> AsyncIterator[RabbitBroker]:
 
 @pytest.fixture
 def rabbit_exchange() -> RabbitExchange:
-    return RabbitExchange("test_exchange")
+    return RabbitExchange("test_exchange", durable=True)
 
 
 async def _assert_call_count(
@@ -256,7 +256,9 @@ async def test_fan_out_exchange_message_delivery(
     handler_1_call_count = Mock()
     handler_2_call_count = Mock()
 
-    fan_out_exchange = RabbitExchange("test_fan_out_exchange", type=ExchangeType.FANOUT)
+    fan_out_exchange = RabbitExchange(
+        "test_fan_out_exchange", type=ExchangeType.FANOUT, durable=True
+    )
 
     @rabbit_broker.subscriber(queue="handler_1", exchange=fan_out_exchange, retry=True)
     async def handler_1(sleep_duration: float) -> None: