🐛 verdi process repair: fix ZMQ broker support (#7350)

Since the ZMQ broker runs inside the daemon and must be stopped for
repair, write revival tasks directly to its on-disk `PersistentQueue`;
the broker picks them up next time the daemon starts. The alternative
would be to spin up a transient broker just to enqueue these tasks,
which adds start/stop ceremony and lifecycle races without changing the
end state on disk.

Also replace RabbitMQ-specific wording in user-facing messages and test
assertions with broker-agnostic terms.

Author: agoscinski

src/aiida/cmdline/commands/cmd_process.py

@@ -518,9 +518,9 @@ def process_repair(manager, broker, dry_run, force):
     N.B.: This command requires the daemon to be stopped.
 
     This command queries the database to find all "active" processes, meaning those that haven't yet reached a terminal
-    state, and cross-references them with the active process tasks in the process queue of RabbitMQ. Any active process
-    that does not have a corresponding process task can be considered a zombie, as it will never be picked up by a
-    daemon worker to complete it and will effectively be "stuck". Any process task that does not correspond to an active
+    state, and cross-references them with the active process tasks in the broker's task queue. Any active process that
+    does not have a corresponding process task can be considered a zombie, as it will never be picked up by a daemon
+    worker to complete it and will effectively be "stuck". Any process task that does not correspond to an active
     process is useless and should be discarded. Finally, duplicate process tasks are also problematic and are discarded.
     """
     from aiida.engine.processes.control import get_active_processes, get_process_tasks, iterate_process_tasks
@@ -579,10 +579,10 @@ def process_repair(manager, broker, dry_run, force):
         echo.echo(set_active_processes.difference(set_process_tasks))
 
     if not state_inconsistent:
-        echo.echo_success('No inconsistencies detected between database and RabbitMQ.')
+        echo.echo_success('No inconsistencies detected between database and broker.')
         return
 
-    echo.echo_warning('Inconsistencies detected between database and RabbitMQ.')
+    echo.echo_warning('Inconsistencies detected between database and broker.')
 
     if dry_run:
         echo.echo_critical('This was a dry-run, no changes will be made.')
@@ -600,11 +600,33 @@ def process_repair(manager, broker, dry_run, force):
             echo.echo_report(f'Acknowledged task `{pid}`')
 
     # Revive zombie processes that no longer have a process task
-    process_controller = manager.get_process_controller()
-    for pid in set_active_processes:
-        if pid not in set_process_tasks:
-            process_controller.continue_process(pid)
-            echo.echo_report(f'Revived process `{pid}`')
+    zombies = set_active_processes.difference(set_process_tasks)
+    if zombies:
+        from aiida.brokers.zmq.broker import ZmqBroker
+
+        if isinstance(broker, ZmqBroker):
+            # For ZMQ, the broker runs inside the daemon (which must be stopped for repair).
+            # Write revival tasks directly to the persistent queue on disk — the broker will
+            # pick them up when the daemon is restarted.  This avoids starting a temporary
+            # broker process and the timing issues that come with it.
+            import uuid
+
+            from plumpy.process_comms import create_continue_body
+
+            from aiida.brokers.zmq.queue import PersistentQueue
+
+            queue_path = broker.storage_path / 'tasks'
+            queue = PersistentQueue(queue_path)
+            for pid in zombies:
+                task_id = uuid.uuid4().hex
+                body = create_continue_body(pid=pid, nowait=True)
+                queue.push(task_id, {'body': body, 'no_reply': True})
+                echo.echo_report(f'Revived process `{pid}`')
+        else:
+            process_controller = manager.get_process_controller()
+            for pid in zombies:
+                process_controller.continue_process(pid)
+                echo.echo_report(f'Revived process `{pid}`')
 
 
 @verdi_process.command('dump')

tests/cmdline/commands/test_process.py

@@ -1021,7 +1021,7 @@ def test_process_repair_consistent(monkeypatch, run_cli_command):
     monkeypatch.setattr(process_control, 'get_process_tasks', lambda *args: [1, 2, 3])
 
     result = run_cli_command(cmd_process.process_repair, use_subprocess=False)
-    assert 'No inconsistencies detected between database and RabbitMQ.' in result.output
+    assert 'No inconsistencies detected between database and broker.' in result.output
 
 
 @pytest.mark.usefixtures('stopped_daemon_client')
@@ -1032,7 +1032,7 @@ def test_process_repair_duplicate_tasks(monkeypatch, run_cli_command):
 
     result = run_cli_command(cmd_process.process_repair, use_subprocess=False)
     assert 'There are duplicates process tasks:' in result.output
-    assert 'Inconsistencies detected between database and RabbitMQ.' in result.output
+    assert 'Inconsistencies detected between database and broker.' in result.output
 
 
 @pytest.mark.usefixtures('stopped_daemon_client')
@@ -1043,7 +1043,7 @@ def test_process_repair_additional_tasks(monkeypatch, run_cli_command):
 
     result = run_cli_command(cmd_process.process_repair, use_subprocess=False)
     assert 'There are process tasks for terminated processes:' in result.output
-    assert 'Inconsistencies detected between database and RabbitMQ.' in result.output
+    assert 'Inconsistencies detected between database and broker.' in result.output
     assert 'Attempting to fix inconsistencies' in result.output
 
 
@@ -1055,7 +1055,7 @@ def test_process_repair_missing_tasks(monkeypatch, run_cli_command):
 
     result = run_cli_command(cmd_process.process_repair, use_subprocess=False)
     assert 'There are active processes without process task:' in result.output
-    assert 'Inconsistencies detected between database and RabbitMQ.' in result.output
+    assert 'Inconsistencies detected between database and broker.' in result.output
     assert 'Attempting to fix inconsistencies' in result.output
 
 
@@ -1066,7 +1066,7 @@ def test_process_repair_dry_run(monkeypatch, run_cli_command):
     monkeypatch.setattr(process_control, 'get_process_tasks', lambda *args: [1, 2])
 
     result = run_cli_command(cmd_process.process_repair, ['--dry-run'], raises=True, use_subprocess=False)
-    assert 'Inconsistencies detected between database and RabbitMQ.' in result.output
+    assert 'Inconsistencies detected between database and broker.' in result.output
     assert 'This was a dry-run, no changes will be made.' in result.output
 
 

tests/cmdline/commands/test_rabbitmq.py

@@ -60,7 +60,7 @@ def test_tasks_analyze_consistent(monkeypatch, run_cli_command):
     monkeypatch.setattr(control, 'get_process_tasks', lambda *args: [1, 2, 3])
 
     result = run_cli_command(cmd_rabbitmq.cmd_tasks_analyze, use_subprocess=False)
-    assert 'No inconsistencies detected between database and RabbitMQ.' in result.output
+    assert 'No inconsistencies detected between database and broker.' in result.output
 
 
 @pytest.mark.usefixtures('stopped_daemon_client')