Use PID file sentinel instead of cmdline matching for broker validation

Write 'aiida-zmq-broker <pid>' to the PID file so is_running() can
validate ownership without fragile command-line string matching.

Also: fix sender type in make_broadcast_message (str, not uuid.UUID),
drop underscore prefix from start/stop_zmq_broker test helpers.

Author: agoscinski

src/aiida/brokers/zmq/broker.py

@@ -87,36 +87,35 @@ def router_endpoint(self) -> str | None:
             return None
         return f'ipc://{sockets_path}/router.sock'
 
+    _PID_SENTINEL = 'aiida-zmq-broker'
+
     def get_service_pid(self) -> int | None:
-        """Read the ZmqBrokerService PID from its PID file."""
+        """Read the ZmqBrokerService PID from its PID file.
+
+        The PID file contains ``aiida-zmq-broker <pid>`` as a sentinel so we
+        can validate ownership without fragile command-line string matching.
+        """
         if not self._service_pid_file.exists():
             return None
         try:
-            return int(self._service_pid_file.read_text().strip())
+            content = self._service_pid_file.read_text().strip()
+            if content.startswith(self._PID_SENTINEL):
+                return int(content.split()[-1])
+            # Fallback: bare PID (old format)
+            return int(content)
         except (ValueError, OSError):
             return None
 
-    def _validate_service_pid(self, pid: int) -> bool:
-        """Check that a PID belongs to a running ZmqBrokerService process.
-
-        Matches both direct invocation (``python -m aiida.brokers.zmq.service``)
-        and the circus-started path (``verdi … daemon broker``).
-        """
-        try:
-            proc = psutil.Process(pid)
-            if proc.is_running() and proc.status() != psutil.STATUS_ZOMBIE:
-                cmdline_str = ' '.join(proc.cmdline())
-                return 'aiida.brokers.zmq' in cmdline_str or 'daemon broker' in cmdline_str
-        except (psutil.NoSuchProcess, psutil.AccessDenied):
-            return False
-        return False
-
     def is_running(self) -> bool:
         """Check if the ZmqBrokerService process is running."""
         pid = self.get_service_pid()
         if pid is None:
             return False
-        return self._validate_service_pid(pid)
+        try:
+            proc = psutil.Process(pid)
+            return proc.is_running() and proc.status() != psutil.STATUS_ZOMBIE
+        except (psutil.NoSuchProcess, psutil.AccessDenied):
+            return False
 
     def get_service_status(self) -> dict[str, t.Any] | None:
         """Read the ZmqBrokerService status from its status file."""

src/aiida/brokers/zmq/protocol.py

@@ -162,7 +162,7 @@ def make_rpc_response(rpc_id: str, sender: str, result: Any = None, error: str |
 
 
 def make_broadcast_message(
-    body: Any, sender: str | uuid.UUID, subject: str | None = None, correlation_id: str | None = None
+    body: Any, sender: str, subject: str | None = None, correlation_id: str | None = None
 ) -> dict:
     """Create a broadcast message dictionary."""
     return {

src/aiida/brokers/zmq/service.py

@@ -75,7 +75,7 @@ def start(self) -> None:
             decoder=YAML_DECODER,
         )
 
-        self._pid_file.write_text(str(os.getpid()))
+        self._pid_file.write_text(f'aiida-zmq-broker {os.getpid()}')
 
         # SIGINT (ctrl-c / circus graceful) + SIGTERM (circus stop)
         signal.signal(signal.SIGINT, self._handle_shutdown)

tests/brokers/test_zmq.py

@@ -16,7 +16,7 @@
 from aiida.brokers.zmq.queue import PersistentQueue
 from aiida.brokers.zmq.server import ZmqBrokerServer
 from aiida.brokers.zmq.service import ZmqBrokerService
-from tests.conftest import _start_zmq_broker, _stop_zmq_broker
+from tests.conftest import start_zmq_broker, stop_zmq_broker
 
 
 class TestZmqDefaults:
@@ -54,48 +54,48 @@ def test_endpoints_no_sockets_file(self, tmp_path):
     def test_start_stop_cycle(self, tmp_path):
         """Test querying a running broker service."""
         broker = ZmqBroker.from_base_path(tmp_path)
-        _start_zmq_broker(broker)
+        start_zmq_broker(broker)
 
         try:
             assert broker.is_running()
             assert broker.get_service_pid() is not None
             assert broker.router_endpoint is not None
         finally:
-            _stop_zmq_broker(broker)
+            stop_zmq_broker(broker)
 
         assert not broker.is_running()
 
     def test_start_already_running(self, tmp_path):
         """Test starting when already running is idempotent."""
         broker = ZmqBroker.from_base_path(tmp_path)
-        _start_zmq_broker(broker)
+        start_zmq_broker(broker)
 
         try:
-            _start_zmq_broker(broker)  # idempotent
+            start_zmq_broker(broker)  # idempotent
             assert broker.is_running()
         finally:
-            _stop_zmq_broker(broker)
+            stop_zmq_broker(broker)
 
     def test_stop_not_running(self, tmp_path):
         """Test stopping when not running is a no-op."""
         broker = ZmqBroker.from_base_path(tmp_path)
-        _stop_zmq_broker(broker)  # Should not raise
+        stop_zmq_broker(broker)  # Should not raise
 
     def test_restart(self, tmp_path):
         """Test restarting the broker service."""
         broker = ZmqBroker.from_base_path(tmp_path)
-        _start_zmq_broker(broker)
+        start_zmq_broker(broker)
 
         try:
             old_pid = broker.get_service_pid()
-            _stop_zmq_broker(broker)
-            _start_zmq_broker(broker)
+            stop_zmq_broker(broker)
+            start_zmq_broker(broker)
             assert broker.is_running()
 
             new_pid = broker.get_service_pid()
             assert new_pid != old_pid
         finally:
-            _stop_zmq_broker(broker)
+            stop_zmq_broker(broker)
 
 
 class TestZmqBrokerServer:
@@ -254,7 +254,7 @@ class TestZmqCommunicator:
     def test_init(self, tmp_path):
         """Test communicator initialization."""
         broker = ZmqBroker.from_base_path(tmp_path)
-        _start_zmq_broker(broker)
+        start_zmq_broker(broker)
 
         try:
             communicator = ZmqCommunicator(
@@ -269,12 +269,12 @@ def test_init(self, tmp_path):
 
             assert communicator.is_closed() is True
         finally:
-            _stop_zmq_broker(broker)
+            stop_zmq_broker(broker)
 
     def test_context_manager(self, tmp_path):
         """Test communicator as context manager."""
         broker = ZmqBroker.from_base_path(tmp_path)
-        _start_zmq_broker(broker)
+        start_zmq_broker(broker)
 
         try:
             with ZmqCommunicator(
@@ -284,7 +284,7 @@ def test_context_manager(self, tmp_path):
 
             assert communicator.is_closed() is True
         finally:
-            _stop_zmq_broker(broker)
+            stop_zmq_broker(broker)
 
 
 class TestZmqBrokerIntegration:
@@ -295,11 +295,11 @@ def zmq_broker(self, tmp_path):
         """Create a ZMQ broker for testing."""
         broker_dir = tmp_path / 'broker' / 'test-uuid'
         broker = ZmqBroker.from_base_path(broker_dir)
-        _start_zmq_broker(broker)
+        start_zmq_broker(broker)
 
         yield broker
 
-        _stop_zmq_broker(broker)
+        stop_zmq_broker(broker)
 
     def test_broker_lifecycle(self, zmq_broker):
         """Test the broker lifecycle."""

tests/conftest.py

@@ -66,7 +66,7 @@ class TestBrokerBackend(Enum):
     NONE = 'none'
 
 
-def _start_zmq_broker(broker, timeout: float = 10.0):
+def start_zmq_broker(broker, timeout: float = 10.0):
     """Start a ZMQ broker service subprocess for testing.
 
     :param broker: A ``ZmqBroker`` instance (has ``base_path``, ``is_running()``, etc.)
@@ -93,13 +93,13 @@ def _start_zmq_broker(broker, timeout: float = 10.0):
     raise TimeoutError(f'ZMQ broker did not start within {timeout}s')
 
 
-def _stop_zmq_broker(broker, timeout: float = 5.0):
+def stop_zmq_broker(broker, timeout: float = 5.0):
     """Stop a ZMQ broker service subprocess for testing.
 
     :param broker: A ``ZmqBroker`` instance
     """
     pid = broker.get_service_pid()
-    if pid is None or not broker._validate_service_pid(pid):
+    if pid is None or not broker.is_running():
         broker._cleanup_stale_service_files()
         return
 
@@ -111,7 +111,7 @@ def _stop_zmq_broker(broker, timeout: float = 5.0):
 
     start_time = time.time()
     while time.time() - start_time < timeout:
-        if not broker._validate_service_pid(pid):
+        if not broker.is_running():
             broker._cleanup_stale_service_files()
             return
         time.sleep(0.1)
@@ -283,11 +283,11 @@ def aiida_profile(pytestconfig, aiida_config, aiida_profile_factory, config_psql
         # Start ZMQ broker service if needed (tests don't use circus)
         broker_instance = get_manager().get_broker()
         if broker_instance is not None and hasattr(broker_instance, 'is_running'):
-            _start_zmq_broker(broker_instance)
+            start_zmq_broker(broker_instance)
             try:
                 yield profile
             finally:
-                _stop_zmq_broker(broker_instance)
+                stop_zmq_broker(broker_instance)
         else:
             yield profile