Print warning when incompatible RabbitMQ version is used (#5315)

The lower limit for the `kiwipy` requirement is upped to `>=0.7.5` since
that provides the `server_properties` property on the `RmqCommunicator`.
This allows us to determine the version of the RabbitMQ server that is
configured for the profile. In `load_profile` a check is added that will
print a warning if this version is greater than 3.7.

For RabbitMQ 3.8 and up, the default configuration will have a default
timeout on channels of around 30 minutes. This means that tasks that are
not acknowledged within that timelimit will be requeued. If the task was
actually still running with a daemon worker, it can be started by a
second runner simultaneously with all sorts of problems as a consequence.

aiida/cmdline/commands/cmd_status.py

@@ -59,6 +59,7 @@ def verdi_status(print_traceback, no_rmq):
     from aiida import __version__
     from aiida.cmdline.utils.daemon import delete_stale_pid_file, get_daemon_status
     from aiida.common.utils import Capturing
+    from aiida.manage.configuration import get_rabbitmq_version, is_rabbitmq_version_supported
     from aiida.manage.configuration.settings import AIIDA_CONFIG_FOLDER
     from aiida.manage.manager import get_manager
 
@@ -138,7 +139,12 @@ def verdi_status(print_traceback, no_rmq):
             print_status(ServiceStatus.ERROR, 'rabbitmq', message, exception=exc, print_traceback=print_traceback)
             exit_code = ExitCode.CRITICAL
         else:
-            print_status(ServiceStatus.UP, 'rabbitmq', f'Connected as {profile.get_rmq_url()}')
+            version = get_rabbitmq_version()
+            connection = f'Connected to RabbitMQ v{version} as {profile.get_rmq_url()}'
+            if is_rabbitmq_version_supported():
+                print_status(ServiceStatus.UP, 'rabbitmq', connection)
+            else:
+                print_status(ServiceStatus.WARNING, 'rabbitmq', 'Incompatible RabbitMQ version detected! ' + connection)
 
     # Getting the daemon status
     try:

aiida/manage/configuration/__init__.py

@@ -59,6 +59,38 @@
 BACKEND_UUID = None  # This will be set to the UUID of the profile as soon as its corresponding backend is loaded
 
 
+def is_rabbitmq_version_supported():
+    """Return whether the version of RabbitMQ configured for the current profile is supported.
+
+    Versions 3.8 and above are not compatible with AiiDA with default configuration.
+
+    :return: boolean whether the current RabbitMQ version is supported.
+    """
+    from packaging.version import parse
+    return get_rabbitmq_version() < parse('3.8')
+
+
+def get_rabbitmq_version():
+    """Return the version of the RabbitMQ server that the current profile connects to.
+
+    :return: :class:`packaging.version.Version`
+    """
+    from packaging.version import parse
+
+    from aiida.manage.manager import get_manager
+    communicator = get_manager().get_communicator()
+    return parse(communicator.server_properties['version'].decode('utf-8'))
+
+
+def check_rabbitmq_version():
+    """Check the version of RabbitMQ that is being connected to and emit warning if the version is not compatible."""
+    from aiida.cmdline.utils import echo
+    if not is_rabbitmq_version_supported():
+        echo.echo_warning(f'RabbitMQ v{get_rabbitmq_version()} is not supported and will cause unexpected problems!')
+        echo.echo_warning('It can cause long-running workflows to crash and jobs to be submitted multiple times.')
+        echo.echo_warning('See https://github.com/aiidateam/aiida-core/wiki/RabbitMQ-version-to-use for details.')
+
+
 def check_version():
     """Check the currently installed version of ``aiida-core`` and warn if it is a post release development version.
 
@@ -121,6 +153,9 @@ def load_profile(profile=None):
     # this function relies on the logging being properly configured for the warning to show.
     check_version()
 
+    # Check whether a compatible version of RabbitMQ is being used.
+    check_rabbitmq_version()
+
     return PROFILE
 
 

environment.yml

@@ -20,7 +20,7 @@ dependencies:
 - ipython~=7.20
 - jinja2~=3.0
 - jsonschema~=3.0
-- kiwipy[rmq]~=0.7.4
+- kiwipy[rmq]~=0.7.5
 - importlib-metadata~=4.3
 - numpy~=1.19
 - pamqp~=2.3

pyproject.toml

@@ -38,7 +38,7 @@ dependencies = [
         "ipython~=7.20",
         "jinja2~=3.0",
         "jsonschema~=3.0",
-        "kiwipy[rmq]~=0.7.4",
+        "kiwipy[rmq]~=0.7.5",
         "importlib-metadata~=4.3",
         "numpy~=1.19",
         "pamqp~=2.3",

requirements/requirements-py-3.10.txt

@@ -60,7 +60,7 @@ jupyter-console==6.4.0
 jupyter-core==4.9.1
 jupyterlab-pygments==0.1.2
 jupyterlab-widgets==1.0.2
-kiwipy==0.7.4
+kiwipy==0.7.5
 kiwisolver==1.3.2
 Mako==1.1.6
 MarkupSafe==2.0.1

requirements/requirements-py-3.8.txt

@@ -61,7 +61,7 @@ jupyter-console==6.4.0
 jupyter-core==4.9.1
 jupyterlab-pygments==0.1.2
 jupyterlab-widgets==1.0.2
-kiwipy==0.7.4
+kiwipy==0.7.5
 kiwisolver==1.3.2
 Mako==1.1.6
 MarkupSafe==2.0.1

requirements/requirements-py-3.9.txt

@@ -60,7 +60,7 @@ jupyter-console==6.4.0
 jupyter-core==4.9.1
 jupyterlab-pygments==0.1.2
 jupyterlab-widgets==1.0.2
-kiwipy==0.7.4
+kiwipy==0.7.5
 kiwisolver==1.3.2
 Mako==1.1.6
 MarkupSafe==2.0.1

tests/conftest.py

@@ -243,6 +243,13 @@ def _create_profile(name='test-profile', **kwargs):
             'database_name': kwargs.pop('database_name', name),
             'database_username': kwargs.pop('database_username', 'user'),
             'database_password': kwargs.pop('database_password', 'pass'),
+            'broker_protocol': kwargs.pop('broker_protocol', 'amqp'),
+            'broker_username': kwargs.pop('broker_username', 'guest'),
+            'broker_password': kwargs.pop('broker_password', 'guest'),
+            'broker_host': kwargs.pop('broker_host', 'localhost'),
+            'broker_port': kwargs.pop('broker_port', 5672),
+            'broker_virtual_host': kwargs.pop('broker_virtual_host', ''),
+            'broker_parameters': kwargs.pop('broker_parameters', {}),
             'repository_uri': f"file:///{os.path.join(repository_dirpath, f'repository_{name}')}",
         }