👌 Unify task timeout into broker.task_timeout config option (#7284)

Replace `rmq.task_timeout` with a broker-agnostic `broker.task_timeout`
option used by both ZMQ and RabbitMQ backends. The old option is kept
with a `deprecated_by` marker on the Field, enabling generic deprecation
handling in `Manager.get_option` and `verdi config set`.

Author: agoscinski

docs/source/howto/installation.rst

@@ -158,7 +158,7 @@ To view all configuration options set for the current profile:
     logging.paramiko_loglevel              default   WARNING
     logging.plumpy_loglevel                default   WARNING
     logging.sqlalchemy_loglevel            default   WARNING
-    rmq.task_timeout                       default   10
+    broker.task_timeout                    default   10
     runner.poll.interval                   profile   50
     transport.task_maximum_attempts        global    6
     transport.task_retry_initial_interval  default   20

docs/source/installation/guide_complete.rst

@@ -142,6 +142,12 @@ Or when using ``verdi profile setup``:
 
     The ``verdi presto`` command automatically falls back to the ZMQ broker when RabbitMQ is not available on the localhost.
 
+.. tip::
+
+    The ``broker.task_timeout`` configuration option controls how long a caller waits for a response from the broker (default: 10 seconds).
+    This option applies to both the ZMQ and RabbitMQ backends.
+    It can be changed with ``verdi config set broker.task_timeout <seconds>``.
+
 .. _installation:guide-complete:rabbitmq:
 
 RabbitMQ

docs/source/internals/broker.rst

@@ -355,10 +355,10 @@ Timeouts
    * - Constant / option
      - Default
      - Purpose
-   * - ``zmq.task_timeout`` |br| (``verdi config``)
+   * - ``broker.task_timeout`` |br| (``verdi config``)
      - 10s
      - How long a caller waits for a task or RPC response from the broker.
-       Fires a ``TimeoutError`` on the pending Future. Analogous to ``rmq.task_timeout``.
+       Fires a ``TimeoutError`` on the pending Future. Replaces the deprecated ``rmq.task_timeout``.
    * - ``BROKER_READY_TIMEOUT``
      - 10s
      - How long ``get_communicator()`` polls for the broker to write its socket files at startup.
@@ -383,7 +383,7 @@ Timeouts
      - 5s
      - How often the broker service writes its status JSON to disk.
 
-Only ``zmq.task_timeout`` is user-configurable.
+Only ``broker.task_timeout`` is user-configurable.
 All other values are developer-tunable constants in ``defaults.py``.
 
 

src/aiida/brokers/rabbitmq/broker.py

@@ -82,7 +82,7 @@ def _create_communicator(self) -> 'RmqThreadCommunicator':
             task_exchange=get_task_exchange_name(self._prefix),
             task_queue=get_launch_queue_name(self._prefix),
             task_prefetch_count=get_config_option('daemon.worker_process_slots'),
-            async_task_timeout=get_config_option('rmq.task_timeout'),
+            async_task_timeout=get_config_option('broker.task_timeout'),
             # This is needed because the verdi commands will call this function and when called in unit tests the
             # testing_mode cannot be set.
             testing_mode=self._profile.is_test_profile,

src/aiida/brokers/zmq/broker.py

@@ -158,9 +158,11 @@ def get_communicator(self, wait_for_broker: float = BROKER_READY_TIMEOUT) -> Zmq
                     msg = f'Broker did not become ready within {wait_for_broker}s: {self}'
                     raise ConnectionError(msg)
 
+            from aiida.manage.configuration import get_config_option
+
             self._communicator = ZmqCommunicator(
                 router_endpoint=router_endpoint,
-                task_timeout=10,
+                task_timeout=get_config_option('broker.task_timeout'),
             )
             self._communicator.start()
 

src/aiida/brokers/zmq/defaults.py

@@ -9,7 +9,8 @@
 """Default constants for the ZMQ broker.
 
 These are **developer-tunable** defaults, not exposed to end users via
-``verdi config``.
+``verdi config``.  User-facing options live in the config schema
+(e.g. ``broker.task_timeout``).
 """
 
 from __future__ import annotations

src/aiida/cmdline/commands/cmd_config.py

@@ -160,6 +160,10 @@ def verdi_config_set(ctx, option, value, globally, append, remove):
     elif option.valid_type == typing.List[str]:
         value = [value]
 
+    # Warn about deprecated option names
+    if option.deprecated_by:
+        echo.echo_warning(f'`{option.name}` is deprecated, use `{option.deprecated_by}` instead.')
+
     # Set the specified option
     try:
         value = config.set_option(option.name, value, scope=scope)

src/aiida/manage/configuration/config.py

@@ -115,7 +115,14 @@ class ProfileOptionsSchema(BaseModel, defer_build=True):
     transport__task_maximum_attempts: int = Field(
         5, description='Maximum number of transport task attempts before a Process is Paused.'
     )
-    rmq__task_timeout: int = Field(10, description='Timeout in seconds for communications with RabbitMQ.')
+    broker__task_timeout: int = Field(
+        10, description='Timeout in seconds for task/RPC communications with the message broker.'
+    )
+    rmq__task_timeout: int = Field(
+        10,
+        description='Timeout in seconds for communications with RabbitMQ.',
+        json_schema_extra={'deprecated_by': 'broker.task_timeout'},
+    )
     storage__sandbox: Optional[str] = Field(
         None, description='Absolute path to the directory to store sandbox folders.'
     )

src/aiida/manage/configuration/options.py

@@ -46,6 +46,11 @@ def default(self) -> Any:
     def description(self) -> str:
         return self._field.description
 
+    @property
+    def deprecated_by(self) -> str | None:
+        """Return the name of the option that replaces this one, or ``None``."""
+        return self._schema.get('deprecated_by')
+
     @property
     def global_only(self) -> bool:
         from .config import ProfileOptionsSchema

src/aiida/manage/configuration/schema/config-v9.schema.json

@@ -143,11 +143,18 @@
           "description": "Toggle whether the profile can be specified in requests submitted to the REST API",
           "global_only": true
         },
+        "broker.task_timeout": {
+          "type": "integer",
+          "default": 10,
+          "minimum": 1,
+          "description": "Timeout in seconds for task/RPC communications with the message broker"
+        },
         "rmq.task_timeout": {
           "type": "integer",
           "default": 10,
           "minimum": 1,
-          "description": "Timeout in seconds for communications with RabbitMQ"
+          "description": "Timeout in seconds for communications with RabbitMQ",
+          "deprecated_by": "broker.task_timeout"
         },
         "storage.sandbox": {
           "type": "string",