diff --git a/benchmark/matched_cellprofiler_batch.py b/benchmark/matched_cellprofiler_batch.py index d4cb5be31..333ec8b21 100644 --- a/benchmark/matched_cellprofiler_batch.py +++ b/benchmark/matched_cellprofiler_batch.py @@ -30,7 +30,6 @@ ensure_global_config_context, rebuild_lazy_config_with_new_global_reference, ) -from zmqruntime import DataControlPortPairAuthority from zmqruntime.messages import ExecutionStatusSnapshot from benchmark.adapters.cellprofiler import ( @@ -114,7 +113,6 @@ from openhcs.interop.cellprofiler.plate_workspace import ( prepare_cellprofiler_input_workspace, ) -from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG from openhcs.runtime.zmq_execution_client import ( OpenHCSExecutionSubmission, ZMQExecutionClient, @@ -1206,11 +1204,7 @@ def main(argv: list[str] | None = None) -> int: if args.preflight_native_references: execution_scope = nullcontext(None) else: - port = DataControlPortPairAuthority.acquire( - OPENHCS_ZMQ_CONFIG, - transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, - ).data_port - execution_scope = ZMQExecutionClient(port=port, persistent=False) + execution_scope = ZMQExecutionClient.with_private_server() receipts = {} with execution_scope as client: for case in selected_cases: diff --git a/benchmark/openhcs_measured_run.py b/benchmark/openhcs_measured_run.py index ef66246f8..8b9ccd536 100644 --- a/benchmark/openhcs_measured_run.py +++ b/benchmark/openhcs_measured_run.py @@ -8,11 +8,14 @@ import sys import time from collections.abc import Callable, Mapping -from dataclasses import dataclass +from contextlib import AbstractContextManager +from dataclasses import ( + dataclass, + field, +) from pathlib import Path from typing import Any -from zmqruntime import DataControlPortPairAuthority from zmqruntime.messages import PongResponse from benchmark.contracts.measured_run_receipt import ( @@ -31,7 +34,6 @@ RuntimeArtifactExecutionObservation, ) from openhcs.runtime.zmq_application import OPENHCS_ENDPOINT_APPLICATION -from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG from openhcs.runtime.zmq_execution_client import ( ZMQExecutionRequestBuilder, OpenHCSExecutionSubmission, @@ -251,6 +253,29 @@ def _require_measured_submission( raise ValueError("Measured OpenHCS observation export path must be absolute.") +def measured_execution_client( + *, + execution_port: int | None, + progress_callback=None, +) -> AbstractContextManager[ZMQExecutionClient]: + """The client of one measured run. + + With no port the run starts a server at an address of its own. A port is + an address the caller named; the client still refuses a server there that + another process started for its own use or that runs other code. + """ + + if execution_port is None: + return ZMQExecutionClient.with_private_server( + progress_callback=progress_callback + ) + return ZMQExecutionClient( + port=execution_port, + persistent=False, + progress_callback=progress_callback, + ) + + def execute_measured_openhcs_pipeline( *, submission: OpenHCSExecutionSubmission, @@ -263,18 +288,9 @@ def execute_measured_openhcs_pipeline( """Measure an ordinary pipeline submission with its requested observation.""" _require_measured_submission(submission, expected_axis_count) - client_port = execution_port - if client_port is None and require_owned_server: - client_port = DataControlPortPairAuthority.acquire( - OPENHCS_ZMQ_CONFIG, - transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, - ).data_port - client = ZMQExecutionClient( - port=client_port, - persistent=False, - progress_callback=timing_observer, - ) - with client: + with measured_execution_client( + execution_port=execution_port, progress_callback=timing_observer + ) as client: return execute_measured_openhcs_pipeline_on_client( client=client, submission=submission, diff --git a/benchmark/well_throughput_scaling.py b/benchmark/well_throughput_scaling.py index a088abe9b..aa2571cef 100644 --- a/benchmark/well_throughput_scaling.py +++ b/benchmark/well_throughput_scaling.py @@ -17,7 +17,6 @@ import psutil from objectstate.lazy_factory import rebuild_lazy_config_with_new_global_reference -from zmqruntime import DataControlPortPairAuthority if TYPE_CHECKING: from benchmark.reports.cppipe_figures import BenchmarkMetricRow @@ -34,6 +33,7 @@ _ZMQProgressTimingObserver, execute_measured_openhcs_pipeline, execute_measured_openhcs_pipeline_on_client, + measured_execution_client, ) from benchmark.timing import BenchmarkPhase, PhaseTimingTrace from openhcs.core.components.parser_metaprogramming import FilenameParseResult @@ -68,7 +68,6 @@ prepare_cellprofiler_input_workspace, ) from openhcs.core.dataset_sources.source_schema import SourceSchemaFilenameParser -from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG from openhcs.runtime.zmq_execution_client import ( OpenHCSExecutionSubmission, ZMQExecutionClient, @@ -2001,16 +2000,9 @@ def run_well_throughput_suite( if observation_key in completed or observation_key in skipped: continue if reuse_execution_server and shared_client is None: - client_port = execution_port - if client_port is None: - client_port = DataControlPortPairAuthority.acquire( - OPENHCS_ZMQ_CONFIG, - transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, - ).data_port shared_client = client_stack.enter_context( - ZMQExecutionClient( - port=client_port, - persistent=False, + measured_execution_client( + execution_port=execution_port, progress_callback=shared_observer, ) ) diff --git a/docs/refactor/consolidation-20261009/B7-server-isolation.md b/docs/refactor/consolidation-20261009/B7-server-isolation.md new file mode 100644 index 000000000..d8a35d7e2 --- /dev/null +++ b/docs/refactor/consolidation-20261009/B7-server-isolation.md @@ -0,0 +1,124 @@ +# B7: an execution client reaches only a server it may use + +**Head audited:** `openhcs` `main` at `625150118` (#1207); `zmqruntime` `main` at `dc327e6` (0.6.1). **Rules:** [00-RULES.md](00-RULES.md). **Origin:** both (the attach rule is zmqruntime's; the shared namespace and the launch are OpenHCS's). +**Shared abstractions.** *Builds:* `zmqruntime.EndpointLaunch` (where a server's code is, which process started it, who may use it) and the `EndpointAudience` family. *Uses:* `AutoRegisterMeta`, `WireMessage`, `ProcessIdentity`, `OpenHCSRuntimeImportAuthority`. + +## What is wrong + +**Nothing relates a client to the server it uses: a client attaches to whatever answers at an address, every OpenHCS process of one user shares one address namespace, and nothing ever stops a server whose starter is gone.** + +How a client picks a server today: + +- **One namespace per user.** `OPENHCS_ZMQ_CONFIG` is a module singleton, the default of 30 parameters. Its endpoint is data port 7777 and control port 8777. On Linux and macOS the transport is IPC, and `IpcTransportDeclaration.socket_path` (`zmqruntime/transport_modes.py:383`) maps a port to `~/.openhcs/ipc/openhcs-zmq-.sock`. Every checkout, worktree and installed copy run by one user resolves the same file for the same port. On Windows the transport is TCP and the namespace is the machine's loopback ports. +- **Attach to whatever answers.** `ZMQClient._connect_locked` (`zmqruntime/client.py:1060`) takes the startup lock, and if the socket file exists it pings the control socket. Any reply with `ready=True` makes it an `AttachedEndpointConnection`. The reply already carries `process_identity` and `application`; neither is read. `connect_existing` does the same. +- **The only identity check is somewhere else and compares too little.** `EndpointApplication` is `(identifier, version)`. `ZMQClientService` (GUI and session) and the function-catalog client compare it after connecting. A plain `ZMQExecutionClient.connect()` (scripts, the benchmark harness, `ExecutionConnectionSpec.execution_client` for MCP) never compares it. Two worktrees have the same identifier and version, so the check passes between them anyway. +- **"A free port" is not reserved.** `DataControlPortPairAuthority.acquire` returns the first pair from 7777 up whose socket files are absent. It reserves nothing. Three harness sites and 9 test files call it and then build a client on that port. +- **Discovery scan.** `ZMQClient.scan_servers` pings a port range (`ports_per_server_type` = 10 per role) in the same namespace and returns every reply. The GUI server browser and `RuntimeServerService.scan` list them; `ZMQServerScanService` is what `test_zmq_browser_scan_service_discovers_ui_bridge_default_transport` exercises, against a `UiBridgeControlServer` bound in the user's real `~/.openhcs/ipc`. +- **UI-bridge descriptors.** The running GUI writes `ui_bridge_*.json` into one per-user runtime directory (`UiBridgeDescriptorDirectoryAuthority`), and `UiBridgeProcessAdvertisedDescriptorCatalog` also reads `/proc/*/environ` of every process for `OPENHCS_UI_BRIDGE_DESCRIPTOR_FILE`. An MCP client with no explicit descriptor takes the only live one, from any checkout. Requests carry `(identifier, version)` and an auth token. This is a client driving a GUI, not choosing an execution server; it is U4's and is recorded under Reassigned. + +The three observed failures are one sequence: + +- The parity test picks one port with `acquire` and keeps it for all 30 workflows (`tests/integration/test_cellprofiler_official30_zmq.py:201`). +- Each workflow builds `ZMQExecutionClient(port=port, persistent=False)`, which starts a server and terminates it at the end of the workflow. Between workflows the socket file is absent. +- A second process (another parity run, or any OpenHCS process on the default port, because `acquire` starts at 7777) sees the port free and starts its own server there. +- The first run's next workflow finds the socket, gets a ready reply and attaches. It now runs on code imported from the other worktree. +- When the other process ends, its server stops under the first run. `benchmark/openhcs_measured_run.py:297` reports "client entered without a connected endpoint" for the same collision seen from the other side. + +`require_owned_server` exists in the harness (`openhcs_measured_run.py:305`) and would have refused; the parity path does not pass it. It stays: a measured run may still be given a port, and then needs a server it started. + +Why a server lived 31 hours: + +- `ExecutionTransportConfig.persistent` defaults to `True`. `ExecutionRuntimeLaunchPlan.spawn` launches a persistent server with `start_new_session=True` (its own session and process group), and `OwnedEndpointConnection.close_client(persistent=True)` does nothing. +- The server holds no reference to the process that started it. `serve_forever` loops until a shutdown request or a signal. There is no idle limit and no check on the starter. +- A non-persistent server is stopped only by its client's `disconnect()`. A client that is killed (a pytest timeout, an agent stopped mid-run) never calls it. +- The code is in memory, so deleting the worktree changes nothing. No stale group is running at this head to inspect; the cause is read from the code. + +## Target + +**The relation, declared once.** A server states its launch, and `ZMQClient` checks that statement at the one place a connection is established. + +```python +class EndpointAudience(ABC, metaclass=AutoRegisterMeta): # who may use a server + def admits(owner, client) -> bool + def orphan_stops(unfinished_work) -> bool +class OwnerOnlyAudience(EndpointAudience): ... # a non-persistent server +class AnySessionAudience(EndpointAudience): ... # a persistent server + +@dataclass(frozen=True) +class EndpointLaunch(WireMessage): + code_root: str # directory the server's top-level package is imported from + owner: ProcessIdentity # the process that started it + audience: str # wire name, derived from the family registry +``` + +- `ZMQServer` holds its `EndpointLaunch` and reports it in every heartbeat (`PongResponse.launch`). +- `code_root()` is a classmethod on clients and servers. Its default is the directory the class's top-level package is imported from. OpenHCS's client and server both inherit `ImportRootEndpoint` (`openhcs/runtime/import_authority.py`), which states `OpenHCSRuntimeImportAuthority.current().import_root`, the root its child processes are already launched from, so a client subclass defined in a script or a test is still this checkout's client. +- **A client that starts a server** hands it the launch (`owner` = the client process, audience from `persistent`) as the launcher's `--launch` argument, which the launcher records with `ZMQServer.launched_by`. A launcher run by hand defaults to a server its own process started. The client connects only if the server that becomes ready reports exactly that launch. +- **A client that attaches** (an occupied endpoint in `connect`, or `connect_existing`) is refused with `EndpointLaunchMismatchError` unless the server reports a launch, its `code_root` equals the client's, and its audience admits the client process. The error names both code roots and the owner. +- The check applies to endpoints on this host. A remote TCP host is an endpoint the user named; its paths and process ids mean nothing here. +- **Namespace.** `ZMQConfig.socket_root()` owns where socket and lock files live. `OpenHCSZMQConfig` derives it from the import root: `~/.openhcs//`. Two checkouts no longer share addresses, so the refusal is the second line, not the normal case. `OPENHCS_SOCKET_ROOT` replaces the root for one process tree; it is read at each call, and child processes inherit it. An absolute `ipc_socket_dir` still selects an explicit directory. +- **Orphans.** The server checks `launch.owner` once a second in its message loop. When the owner is gone, an owner-only server cancels its work and stops; an any-session server stops once it has no running or queued execution. +- **Tests and harness.** `tests/conftest.py` calls `use_private_socket_root()`, which makes a new directory beneath the checkout's socket root and exports it as `OPENHCS_SOCKET_ROOT` (one per pytest process, so per xdist worker). The directory is not under `TMPDIR`: a socket path longer than `zmq.IPC_PATH_MAX_LEN` (107) cannot be bound, and a run with a deep `TMPDIR` put every server there. The `unix_socket_dir` fixture is a subdirectory of the same root. +- **A path no server can bind.** `ZMQExecutionClient.connect` raises `SocketPathTooLongError` (path, length, limit) before it starts a server. A session whose compile cannot connect publishes `CompilationFailed` per dataset, so a caller waiting for the compilation (the measured CLI) ends instead of waiting. The harness uses `ZMQExecutionClient.with_private_server`, which takes its address from `TransportDeclaration.private_endpoint` (a new socket directory for IPC, the first unbound pair for TCP) and raises if the server does not start. `benchmark.openhcs_measured_run.measured_execution_client` is the one place the harness builds a client: private with no port, the named port otherwise. + +Identity that must match: code location and owning process (through the audience). Version stays where it is: `EndpointApplication`, compared by `ZMQClientService`. A server edited after it started has the same root and version and is not detected; a source digest would be needed and no surface owns that. + +Deleted (all **derived**: `private_endpoint` and `socket_root` now own the fact): + +- the three `DataControlPortPairAuthority.acquire` + `ZMQExecutionClient(port=...)` pairs in the harness; +- the execution port the three official30 tests chose (`_free_zmq_port_pair` stays for viewer ports); +- the hand-built `Path.home() / ".openhcs" / "ipc"` in `viewer_endpoint_discovery.py`. + +Behaviour removed: a measured run with no port and without `require_owned_server` no longer uses the shared default port 7777; it starts a private server (**contract**: this surface's ownership rule). + +## Persisted state + +Runtime only. Socket and lock files move from `~/.openhcs/ipc` and `~/.openhcs/tcp` to `~/.openhcs//`; servers started before the upgrade are not found and are not reused. The saved `OpenHCSZMQConfig` preference keeps every field. + +## Guards + +`tests/unit/test_b7_server_isolation_guards.py`: + +- Every `ZMQClient` method that assigns a connection calls a launch check (AST). +- No ZMQ endpoint code (`zmqruntime.transport_modes`, `zmqruntime.client`, `openhcs/{runtime,agent,core,pyqt_gui}`) builds a socket path from `Path.home()`. +- No module in `benchmark/` names `DataControlPortPairAuthority`. +- Client and server state the import root as their code root, including a subclass defined elsewhere. +- Each checkout has its own socket root, and the test run's root is private. +- A child process computes its parent's socket root, code root and audience; a private root can be bound whatever `TMPDIR` is; a client refuses a path too long to bind without starting a server. + +## Tests + +- zmqruntime `tests/test_endpoint_launch.py`, with real server processes: two servers from two code roots, each client reaches only its own, and a client aimed at the other's address is refused with both roots in the message; an owner-only server refuses another process and a persistent one admits it; a server that reports no launch is refused; a server exits after its owner is killed (both audiences); a persistent orphan finishes its work first. +- OpenHCS `tests/integration/test_execution_server_isolation.py`, with the real launcher: the started server reports the launch its client handed it; a client of another checkout and a client of another process are refused at the same address and the server is unharmed; the server and its children exit after the owner is killed. +- Both orphan tests and the two-root test were run against the code with the check removed and fail there. + +## New-case experiments + +A new kind of server (a viewer, a bridge) that must not be shared across checkouts: today it would need its own check after connecting. After: it subclasses `ZMQServer` and its clients subclass `ZMQClient`; nothing else. + +## Done when + +- A client cannot hold a connection to a local server whose launch it did not verify. +- A server whose owner is gone exits without a request. +- The parity check passes 30/30 with no port chosen by the test. +- The guards pass; the structural ratchet and nra-r1 show no regression. + +## Reassigned + +| Item | Evidence | Belongs to | +|---|---|---| +| UI-bridge descriptor discovery takes the only live descriptor from any checkout | `ui_bridge_service.py:1563-1585`, `/proc` environment scan at `:1599` | U4 | +| The MCP resident-server socket restates the checkout digest and `~/.openhcs/ipc` | `openhcs/mcp/socket.py:48,61`; it should read `OPENHCS_ZMQ_CONFIG.socket_root()` | U4 | +| `connect` kills the listener of an unresponsive TCP address before starting its own server | `zmqruntime/client.py` `_kill_processes_on_port`; the process may be another session's | zmqruntime (L3) | +| The application version is compared after connecting, and only by `ZMQClientService` and the catalog client | `authoring/session/execution_client.py:202`; a plain client attaches to a same-root server of another version | U1 | +| `IpcTransportDeclaration.private_endpoint` makes its directory under `TMPDIR`, not under `config.socket_root()`; with a deep `TMPDIR` a private server is refused by the length check | `zmqruntime/transport_modes.py` `private_endpoint` | zmqruntime | +| `viewer_endpoint_discovery.py:212` names an undefined `ViewerEndpointDiscoveryResult` | ruff F821 at this head | U4 | + +## Result + +- Production (`openhcs/`): -12 +102. The additions are the socket root, the launch argument, `with_private_server` and `ImportRootEndpoint`; nothing existed to delete, because the relation was missing. zmqruntime `src`: -5 +331. +- Guards: `tests/unit/test_b7_server_isolation_guards.py`, 6 of 6 tests pass. +- Certificates: the three harness `acquire` + client pairs, the test-chosen execution port and the hand-built socket directory are **derived** (`private_endpoint`, `socket_root`). The shared default port for a measured run with no port is **contract** (this surface's rule). No **tests-only** deletion. +- Files of in-flight surfaces touched: `openhcs/agent/services/viewer_endpoint_discovery.py` (U4, one expression); `openhcs/runtime/zmq_execution_client.py`, `zmq_execution_server.py`, `zmq_execution_server_launcher.py` (near K7); `benchmark/openhcs_measured_run.py`, `well_throughput_scaling.py`, `matched_cellprofiler_batch.py` (near K1 and K8, client construction only). + +Release order: merge OpenHCSDev/ZMQRuntime#32 and release zmqruntime 0.6.3 (0.6.2 went to the Windows startup-lock fix); point `external/zmqruntime` at that merge commit; merge this PR. diff --git a/external/zmqruntime b/external/zmqruntime index 482654929..beea6b5e3 160000 --- a/external/zmqruntime +++ b/external/zmqruntime @@ -1 +1 @@ -Subproject commit 4826549299f69ac56e3dfc858d15ba92c8574f95 +Subproject commit beea6b5e3d8b9c60dfe3ba773dba975c473f0036 diff --git a/openhcs/agent/services/viewer_endpoint_discovery.py b/openhcs/agent/services/viewer_endpoint_discovery.py index 538d268e1..1e7dbab41 100644 --- a/openhcs/agent/services/viewer_endpoint_discovery.py +++ b/openhcs/agent/services/viewer_endpoint_discovery.py @@ -53,11 +53,7 @@ def as_dict(self) -> dict[str, Any]: def viewer_ipc_directory() -> Path: """Return the OpenHCS IPC socket directory for the configured transport.""" - return ( - Path.home() - / f".{OPENHCS_ZMQ_CONFIG.app_name}" - / OPENHCS_ZMQ_CONFIG.ipc_socket_dir - ) + return OPENHCS_ZMQ_CONFIG.socket_root() / OPENHCS_ZMQ_CONFIG.ipc_socket_dir def candidate_viewer_ports() -> tuple[int, ...]: diff --git a/openhcs/authoring/session/compile_batch.py b/openhcs/authoring/session/compile_batch.py index 003deaf30..ba3b2ec14 100644 --- a/openhcs/authoring/session/compile_batch.py +++ b/openhcs/authoring/session/compile_batch.py @@ -48,7 +48,14 @@ async def compile_datasets(self, scope_ids: tuple[str, ...]) -> None: session.progress_tracker.clear_execution(previous_execution_id) session.mark_compile_pending(scope_ids) try: - client = await session.connect_client() + try: + client = await session.connect_client() + except Exception as error: + # Without a server no dataset compiles: say so per dataset, so + # whoever waits for the compilation stops waiting. + for scope_id in scope_ids: + self._compile_failed(scope_id, error) + raise session.publish( StatusReported(f"Queueing compilation for {len(scope_ids)} dataset(s)...") ) diff --git a/openhcs/runtime/import_authority.py b/openhcs/runtime/import_authority.py index 17916eeac..4cc50df0a 100644 --- a/openhcs/runtime/import_authority.py +++ b/openhcs/runtime/import_authority.py @@ -67,3 +67,16 @@ def module_process_arguments(self, module_name: str) -> tuple[str, str]: """Project one module launch to interpreter arguments.""" return "-c", self.module_bootstrap_code(module_name) + + +class ImportRootEndpoint: + """A ZMQ client or server whose code root is the OpenHCS import root. + + Both sides of an execution endpoint inherit this, so the root a server + reports and the root a client expects are one declaration: the root every + OpenHCS child process is launched from. + """ + + @classmethod + def code_root(cls) -> str: + return str(OpenHCSRuntimeImportAuthority.current().import_root) diff --git a/openhcs/runtime/zmq_config.py b/openhcs/runtime/zmq_config.py index 4c93c6a69..1f448cb66 100644 --- a/openhcs/runtime/zmq_config.py +++ b/openhcs/runtime/zmq_config.py @@ -2,11 +2,75 @@ from __future__ import annotations +import atexit +import hashlib +import os +import shutil +import tempfile from dataclasses import dataclass +from pathlib import Path +import zmq from zmqruntime.config import NonBlankString from zmqruntime.execution.config import ExecutionTransportConfig +from openhcs.runtime.import_authority import OpenHCSRuntimeImportAuthority + +SOCKET_ROOT_ENV = "OPENHCS_SOCKET_ROOT" +"""Directory that replaces the socket root for one process tree. + +Child processes inherit it, so a test run or a script and every server and +viewer it starts share one namespace that nothing else on the machine sees. +""" + + +def checkout_socket_root() -> Path: + """Socket root of this OpenHCS checkout or installation. + + Port ``n`` names one socket file per root. Deriving the root from the + import root keeps two checkouts run by one user from sharing addresses. + """ + + import_root = OpenHCSRuntimeImportAuthority.current().import_root + digest = hashlib.sha256(str(import_root).encode("utf-8")).hexdigest()[:8] + return Path.home() / ".openhcs" / digest + + +def use_private_socket_root() -> Path: + """Give this process and its children a socket root of their own. + + The root is a new directory beneath the checkout's socket root, so its + length does not depend on ``TMPDIR``: a socket path longer than + ``zmq.IPC_PATH_MAX_LEN`` cannot be bound. It is removed at exit. + """ + + parent = checkout_socket_root() + parent.mkdir(parents=True, exist_ok=True) + root = Path(tempfile.mkdtemp(prefix="run-", dir=parent)) + os.environ[SOCKET_ROOT_ENV] = str(root) + atexit.register(shutil.rmtree, root, ignore_errors=True) + return root + + +class SocketPathTooLongError(ValueError): + """An IPC socket path that no server can bind.""" + + +def require_bindable_address(url: str) -> None: + """Refuse an IPC address the operating system cannot bind. + + A server asked to bind such a path fails at startup, and its client would + only learn that the server never became ready. + """ + + scheme, _, path = url.partition("://") + if scheme == "ipc" and len(os.fsencode(path)) > zmq.IPC_PATH_MAX_LEN: + raise SocketPathTooLongError( + f"Socket path {path} is {len(os.fsencode(path))} bytes; the limit " + f"is {zmq.IPC_PATH_MAX_LEN}. Set {SOCKET_ROOT_ENV} to a shorter " + "directory (a private server takes its directory from TMPDIR)." + ) + @dataclass(frozen=True, slots=True) class OpenHCSZMQConfig(ExecutionTransportConfig): @@ -18,5 +82,9 @@ class OpenHCSZMQConfig(ExecutionTransportConfig): app_name: NonBlankString = "openhcs" """OpenHCS application namespace used in transport identities and runtime paths.""" + def socket_root(self) -> Path: + declared = os.environ.get(SOCKET_ROOT_ENV) + return Path(declared) if declared else checkout_socket_root() + OPENHCS_ZMQ_CONFIG = OpenHCSZMQConfig() diff --git a/openhcs/runtime/zmq_execution_client.py b/openhcs/runtime/zmq_execution_client.py index ed2ab830b..7d2a8e345 100644 --- a/openhcs/runtime/zmq_execution_client.py +++ b/openhcs/runtime/zmq_execution_client.py @@ -7,10 +7,10 @@ import subprocess import sys import time -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Callable, Iterator, Mapping, Sequence from concurrent.futures import CancelledError from copy import deepcopy -from contextlib import AbstractContextManager, nullcontext +from contextlib import AbstractContextManager, contextmanager, nullcontext from dataclasses import dataclass, replace from enum import Enum from pathlib import Path @@ -39,6 +39,7 @@ ControlMessageType, ControlRequestHeader, EndpointApplicationCompatibilityError, + EndpointLaunch, ExecuteRequest, ExecutionRecord, PongResponse, @@ -66,8 +67,15 @@ ) from openhcs.core.steps.function_step import FunctionStep from openhcs.core.xdg_paths import get_openhcs_data_dir, get_openhcs_log_dir -from openhcs.runtime.import_authority import OpenHCSRuntimeImportAuthority -from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG, OpenHCSZMQConfig +from openhcs.runtime.import_authority import ( + ImportRootEndpoint, + OpenHCSRuntimeImportAuthority, +) +from openhcs.runtime.zmq_config import ( + OPENHCS_ZMQ_CONFIG, + OpenHCSZMQConfig, + require_bindable_address, +) from openhcs.runtime.zmq_execution_signature import ( ZMQAuxiliaryExecutionParams, ZMQExecutionCompileControl, @@ -157,6 +165,7 @@ def spawn( config: OpenHCSZMQConfig, *, persistent: bool, + launch: EndpointLaunch, ) -> subprocess.Popen: """Materialize this admitted plan through the canonical process policy. @@ -191,6 +200,8 @@ def spawn( str(self.startup_status_file), "--config-source", _pycodify_config_source(server_config), + "--launch", + launch.to_argument(), ] root_logger = logging.getLogger() current_log_level = root_logger.getEffectiveLevel() @@ -950,7 +961,7 @@ def _send_function_catalog_exchange( ) -class ZMQExecutionClient(FunctionCatalogExecutionClient): +class ZMQExecutionClient(ImportRootEndpoint, FunctionCatalogExecutionClient): """ZMQ client for OpenHCS pipeline execution with progress streaming.""" config: OpenHCSZMQConfig @@ -980,6 +991,35 @@ def __init__( connection_status_callback=connection_status_callback, ) + @classmethod + @contextmanager + def with_private_server( + cls, + *, + progress_callback=None, + config: OpenHCSZMQConfig = OPENHCS_ZMQ_CONFIG, + ) -> Iterator[ZMQExecutionClient]: + """A connected client on a server it started at an address of its own. + + The server stops when the block ends. The address comes from the + transport, so no other process is given it: a run that needs its own + server uses this instead of picking a free port. + """ + + private_endpoint = config.transport_mode.declaration.private_endpoint + with private_endpoint(config) as private_config: + client = cls( + persistent=False, + progress_callback=progress_callback, + config=private_config, + ) + if not client.connect(): + raise RuntimeError("The private execution server did not start") + try: + yield client + finally: + client.disconnect() + def connect( self, timeout: float | None = None, @@ -988,6 +1028,11 @@ def connect( ): """Connect using the OpenHCS endpoint's declared startup deadline.""" + for url in ( + self.endpoint.data_url(self.config), + self.endpoint.control_url(self.config), + ): + require_bindable_address(url) return super().connect( timeout=( self.config.client_connect_timeout_seconds @@ -1279,7 +1324,12 @@ def _spawn_server_process(self) -> subprocess.Popen: plan = self.runtime_launch_plan() self._runtime_launch_plan = None self._startup_status_path = plan.startup_status_file - return plan.spawn(self.endpoint, self.config, persistent=self.persistent) + return plan.spawn( + self.endpoint, + self.config, + persistent=self.persistent, + launch=self.launch(), + ) def runtime_launch_plan(self) -> ExecutionRuntimeLaunchPlan: """Resolve owner defaults without creating directories or warming.""" diff --git a/openhcs/runtime/zmq_execution_server.py b/openhcs/runtime/zmq_execution_server.py index b7c2bd472..1da59560d 100644 --- a/openhcs/runtime/zmq_execution_server.py +++ b/openhcs/runtime/zmq_execution_server.py @@ -33,6 +33,7 @@ from openhcs.core.progress import ProgressEvent, ProgressExecutionContext from openhcs.core.steps.function_step import FunctionStep from openhcs.runtime.environment_provenance import RuntimeEnvironmentSnapshot +from openhcs.runtime.import_authority import ImportRootEndpoint from openhcs.runtime.zmq_application import OPENHCS_ENDPOINT_APPLICATION from openhcs.runtime.zmq_compilation import ( ZMQCompilationRequest, @@ -161,7 +162,7 @@ def stop(self) -> None: self._function_catalog_preparation.cancel_and_join() -class ZMQExecutionServer(FunctionCatalogExecutionServer): +class ZMQExecutionServer(ImportRootEndpoint, FunctionCatalogExecutionServer): """OpenHCS-specific execution server.""" _server_type = "execution" diff --git a/openhcs/runtime/zmq_execution_server_launcher.py b/openhcs/runtime/zmq_execution_server_launcher.py index 477fd88c7..4e46f02f6 100644 --- a/openhcs/runtime/zmq_execution_server_launcher.py +++ b/openhcs/runtime/zmq_execution_server_launcher.py @@ -12,12 +12,15 @@ from pathlib import Path from zmqruntime.config import TransportMode +from zmqruntime.messages import AnySessionAudience, EndpointLaunch from zmqruntime.startup import ( EndpointStartupPhase, EndpointStartupStatusWriter, ) from zmqruntime.transport import TransportEndpoint +from openhcs.runtime.import_authority import ImportRootEndpoint + logger = logging.getLogger(__name__) @@ -69,6 +72,14 @@ def main( action="store_true", help="Prepare endpoint-owned capability caches without binding sockets", ) + parser.add_argument( + "--launch", + type=EndpointLaunch.from_argument, + default=EndpointLaunch.by_this_process( + ImportRootEndpoint.code_root(), AnySessionAudience + ), + help="Launch record of the starting client (default: started by hand)", + ) parser.add_argument( "--startup-status-path", type=str, @@ -192,7 +203,7 @@ def main( server = execution_server_type( log_file_path=args.log_file_path, config=config, - ) + ).launched_by(args.launch) status_reporter.emit( EndpointStartupPhase.PREPARING_CAPABILITIES, diff --git a/pyproject.toml b/pyproject.toml index 7abb8dfaa..d59fe49fe 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -102,7 +102,7 @@ dependencies = [ "psutil>=5.9.0", # External dependencies (PyPI versions for production builds) - "zmqruntime>=0.6.2,<0.7", + "zmqruntime>=0.6.3,<0.7", "pycodify>=0.2.1,<0.3", "objectstate>=2.1.1,<3", "python-introspect>=0.2.8,<0.3", diff --git a/tests/conftest.py b/tests/conftest.py index 4171a9b1c..59ed8b365 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -12,6 +12,13 @@ ensure_source_checkout_external_paths() +from openhcs.runtime.zmq_config import OPENHCS_ZMQ_CONFIG, use_private_socket_root + +# Every socket and lock file of this pytest process, and of the servers and +# viewers it starts, lives under a root no other run sees. Each xdist worker +# imports this module, so each has its own. +use_private_socket_root() + import psutil import pytest from zmqruntime.messages import AckReturnRoute, ProcessIdentity @@ -99,12 +106,14 @@ def cleanup_test_runtime_resources(): @pytest.fixture def unix_socket_dir(): - """A short private directory for AF_UNIX/IPC sockets. + """A private directory for AF_UNIX/IPC sockets, beneath the socket root. ``sun_path`` holds about 104-108 bytes, so a socket under a deep ``tmp_path`` (long ``--basetemp`` or ``TMPDIR``) cannot bind. """ - with tempfile.TemporaryDirectory(prefix="ohcs-sock-") as directory: + with tempfile.TemporaryDirectory( + prefix="s-", dir=OPENHCS_ZMQ_CONFIG.socket_root() + ) as directory: yield Path(directory) diff --git a/tests/integration/test_cellprofiler_official30_zmq.py b/tests/integration/test_cellprofiler_official30_zmq.py index ced039766..63aa3141f 100644 --- a/tests/integration/test_cellprofiler_official30_zmq.py +++ b/tests/integration/test_cellprofiler_official30_zmq.py @@ -201,10 +201,6 @@ def _assert_manifest_reference_parity( persistent=False, ), ) - execution_port = _free_zmq_port_pair( - set(), - transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, - ) observations = run_comparison_suite( cases, output_root=output_root, @@ -214,7 +210,6 @@ def _assert_manifest_reference_parity( openhcs_global_config=global_config, discard_openhcs_outputs=True, continue_on_error=True, - openhcs_execution_port=execution_port, ) assert len(observations) == expected_case_count @@ -592,10 +587,6 @@ def test_official30_nonpersistent_napari_isolated_per_case( native_reference_root = _native_reference_root() assert len(_OFFICIAL30_CASES) == 30 excluded_ports: set[int] = set() - execution_port = _free_zmq_port_pair( - excluded_ports, - transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, - ) viewer_transport_mode = TransportMode.default() viewer_port = _free_zmq_port_pair( excluded_ports, @@ -651,7 +642,6 @@ def test_official30_nonpersistent_napari_isolated_per_case( native_reference_root=native_reference_root, require_native_reference=True, openhcs_global_config=global_config, - openhcs_execution_port=execution_port, discard_openhcs_outputs=False, continue_on_error=True, ) @@ -742,10 +732,6 @@ def test_official30_persistent_fiji_variants_isolated_per_case( assert len(_OFFICIAL30_CASES) == 30 excluded_ports: set[int] = set() - execution_port = _free_zmq_port_pair( - excluded_ports, - transport_mode=OPENHCS_ZMQ_CONFIG.transport_mode, - ) ports_by_viewer = { viewer: _free_zmq_port_pair( excluded_ports, @@ -799,7 +785,6 @@ def test_official30_persistent_fiji_variants_isolated_per_case( native_reference_root=native_reference_root, require_native_reference=True, openhcs_global_config=global_config, - openhcs_execution_port=execution_port, discard_openhcs_outputs=False, continue_on_error=True, ) diff --git a/tests/integration/test_execution_server_isolation.py b/tests/integration/test_execution_server_isolation.py new file mode 100644 index 000000000..58d4bd856 --- /dev/null +++ b/tests/integration/test_execution_server_isolation.py @@ -0,0 +1,108 @@ +"""A real execution client reaches only a server it may use (B7).""" + +from __future__ import annotations + +import subprocess +import sys +from dataclasses import replace + +import psutil +import pytest +from zmqruntime.messages import ( + EndpointLaunch, + EndpointLaunchMismatchError, + OwnerOnlyAudience, + ProcessIdentity, +) + +from openhcs.runtime.import_authority import OpenHCSRuntimeImportAuthority +from openhcs.runtime.zmq_execution_client import ZMQExecutionClient + +SERVER_START_TIMEOUT_SECONDS = 600 +ANOTHER_PROCESS = ProcessIdentity(pid=1, create_time=0.0) + + +class _ClientOfAnotherCheckout(ZMQExecutionClient): + def launch(self) -> EndpointLaunch: + return replace(super().launch(), code_root="/another/checkout") + + +class _ClientOfAnotherProcess(ZMQExecutionClient): + def launch(self) -> EndpointLaunch: + return replace(super().launch(), owner=ANOTHER_PROCESS) + + +@pytest.fixture(scope="module") +def private_client(): + with ZMQExecutionClient.with_private_server() as client: + yield client + + +def test_a_started_server_reports_the_launch_its_client_handed_it(private_client): + launch = private_client.connected_endpoint.launch + + assert launch == private_client.launch() + assert launch.code_root == str( + OpenHCSRuntimeImportAuthority.current().import_root + ) + assert launch.owner == ProcessIdentity.current() + assert launch.audience_type is OwnerOnlyAudience + assert private_client.owned_server_process_is_alive() is True + + +@pytest.mark.parametrize( + ("stray_type", "named_in_refusal"), + [ + (_ClientOfAnotherCheckout, "/another/checkout"), + (_ClientOfAnotherProcess, "for its own use"), + ], +) +def test_a_client_from_elsewhere_is_refused_at_the_same_address( + private_client, stray_type, named_in_refusal +): + stray = stray_type(persistent=False, config=private_client.config) + + with pytest.raises(EndpointLaunchMismatchError) as refused: + stray.connect(timeout=10) + + assert named_in_refusal in str(refused.value) + assert not stray.is_connected() + assert private_client.owned_server_process_is_alive() is True + assert private_client.endpoint.ping(private_client.config) is not None + + +_OWNER_SCRIPT = """ +import sys, time +from openhcs._source_dependencies import ensure_source_checkout_external_paths +ensure_source_checkout_external_paths() +from openhcs.runtime.zmq_execution_client import ZMQExecutionClient +with ZMQExecutionClient.with_private_server() as client: + print(client.connected_endpoint.process_identity.pid, flush=True) + time.sleep(3600) +""" + + +@pytest.mark.timeout(SERVER_START_TIMEOUT_SECONDS + 60) +def test_an_execution_server_exits_after_its_owner_is_killed(reap_spawned_processes): + import_root = str(OpenHCSRuntimeImportAuthority.current().import_root) + owner = subprocess.Popen( + [ + sys.executable, + "-c", + f"import sys; sys.path.insert(0, {import_root!r})" + _OWNER_SCRIPT, + ], + stdout=subprocess.PIPE, + text=True, + ) + try: + server = psutil.Process(int(owner.stdout.readline())) + workers_and_server = [server, *server.children(recursive=True)] + assert server.is_running() + + owner.kill() + owner.wait(timeout=10) + + _gone, alive = psutil.wait_procs(workers_and_server, timeout=30) + assert alive == [] + finally: + owner.kill() diff --git a/tests/unit/agent/test_owned_runtime_bootstrap.py b/tests/unit/agent/test_owned_runtime_bootstrap.py index bdee8cdb7..37af888f2 100644 --- a/tests/unit/agent/test_owned_runtime_bootstrap.py +++ b/tests/unit/agent/test_owned_runtime_bootstrap.py @@ -21,7 +21,9 @@ from openhcs.agent.path_policy import AgentPathPolicy, AgentPathPolicyError from openhcs.agent.services.runtime_server_service import RuntimeServerService from openhcs.mcp.context import OpenHCSAgentContext -from openhcs.runtime.zmq_config import OpenHCSZMQConfig +from zmqruntime.messages import EndpointAudience, EndpointLaunch + +from openhcs.runtime.zmq_config import SOCKET_ROOT_ENV, OpenHCSZMQConfig from openhcs.runtime.zmq_execution_client import ( ExecutionRuntimeLaunchPlan, ZMQExecutionClient, @@ -55,6 +57,7 @@ def stop(self, timeout=5, kill_timeout=2): @pytest.fixture def setup(tmp_path, monkeypatch): monkeypatch.setenv("HOME", str(tmp_path)) + monkeypatch.delenv(SOCKET_ROOT_ENV) monkeypatch.setenv("XDG_DATA_HOME", str(tmp_path / "data")) monkeypatch.setenv("XDG_CACHE_HOME", str(tmp_path / "cache")) policy = AgentPathPolicy.with_roots( @@ -160,7 +163,15 @@ def test_admitted_plan_owns_materialization_and_child_configuration( module.MemoryType, "subprocess_environment", lambda: environment ) - assert plan.spawn(endpoint, config, persistent=persistent) is native_process + launch_record = EndpointLaunch.by_this_process( + "/code", EndpointAudience.of_persistent(persistent) + ) + assert ( + plan.spawn(endpoint, config, persistent=persistent, launch=launch_record) + is native_process + ) + command = launch.call_args.args[0] + assert command[command.index("--launch") + 1] == launch_record.to_argument() intercepted_spawn.assert_not_called() launch.assert_called_once() policy_selection.assert_called_once_with(detached=persistent) @@ -203,7 +214,8 @@ def test_client_consumes_exact_admitted_plan_on_spawn_failure(setup, monkeypatch calls = [] - def consume(selected_plan, endpoint, config, *, persistent): + def consume(selected_plan, endpoint, config, *, persistent, launch): + assert launch == client.launch() calls.append((selected_plan, endpoint, config, persistent)) raise OSError("controlled no-child spawn failure") @@ -657,6 +669,7 @@ def test_close_native_transport_write_admission_precedes_lifecycle( new_home = tmp_path / "not-authorised" new_home.mkdir() monkeypatch.setenv("HOME", str(new_home)) + monkeypatch.delenv(SOCKET_ROOT_ENV, raising=False) policy = AgentPathPolicy.with_roots( readable_roots=(tmp_path,), writable_roots=(tmp_path / "data", tmp_path / "cache", tmp_path / ".openhcs"), diff --git a/tests/unit/pyqt_gui/test_session_batch_workflows.py b/tests/unit/pyqt_gui/test_session_batch_workflows.py index a715c3178..778d9a741 100644 --- a/tests/unit/pyqt_gui/test_session_batch_workflows.py +++ b/tests/unit/pyqt_gui/test_session_batch_workflows.py @@ -241,9 +241,19 @@ def test_compile_connection_failure_releases_only_its_reserved_dataset( async def connect(): raise RuntimeError("controlled connection failure") + failures = [] + session.subscribe( + lambda record: failures.append(record.event) + if isinstance(record.event, CompilationFailed) + else None + ) monkeypatch.setattr(session, "connect_client", connect) with pytest.raises(RuntimeError, match="controlled connection failure"): asyncio.run(session.compile_datasets((other,))) + # A caller waiting for this compilation is told it ended. + (failure,) = failures + assert failure.scope_id == other + assert "controlled connection failure" in failure.message assert not session.compile_pending assert session.batch.execution_id(running) == "original-run" diff --git a/tests/unit/test_b7_server_isolation_guards.py b/tests/unit/test_b7_server_isolation_guards.py new file mode 100644 index 000000000..33812b1cd --- /dev/null +++ b/tests/unit/test_b7_server_isolation_guards.py @@ -0,0 +1,213 @@ +"""Guards for B7: the checks that keep a client on a server it may use.""" + +from __future__ import annotations + +import ast +import inspect +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest +import zmq + +import zmqruntime.client +import zmqruntime.transport_modes +from zmqruntime.client import ZMQClient +from zmqruntime.config import ZMQConfig + +import openhcs +from openhcs.runtime.import_authority import OpenHCSRuntimeImportAuthority +from openhcs.runtime.zmq_config import ( + OPENHCS_ZMQ_CONFIG, + SOCKET_ROOT_ENV, + SocketPathTooLongError, + checkout_socket_root, + require_bindable_address, +) +from openhcs.runtime.zmq_execution_client import ZMQExecutionClient +from openhcs.runtime.zmq_execution_server import ZMQExecutionServer + +REPOSITORY_ROOT = Path(openhcs.__file__).resolve().parents[1] +LAUNCH_CHECKS = {"_require_started_endpoint", "_require_usable_endpoint"} + + +def _functions_assigning_a_connection() -> dict[str, ast.FunctionDef]: + tree = ast.parse(inspect.getsource(ZMQClient)) + assigning = {} + for function in ast.walk(tree): + if not isinstance(function, ast.FunctionDef): + continue + for node in ast.walk(function): + if not isinstance(node, ast.Assign): + continue + targets = [ast.unparse(target) for target in node.targets] + cleared = isinstance(node.value, ast.Constant) and node.value.value is None + if "self._connection" in targets and not cleared: + assigning[function.name] = function + return assigning + + +def test_a_client_holds_a_connection_only_after_the_launch_check(): + assigning = _functions_assigning_a_connection() + + assert assigning, "ZMQClient no longer assigns _connection where this guard looks" + for name, function in assigning.items(): + if name == "__init__": + continue + called = { + node.func.attr + for node in ast.walk(function) + if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) + } + assert called & LAUNCH_CHECKS, ( + f"ZMQClient.{name} establishes a connection without checking " + "the server's launch" + ) + + +def test_socket_paths_come_from_the_socket_root(): + for module in (zmqruntime.transport_modes, zmqruntime.client): + assert "Path.home()" not in inspect.getsource(module), module.__name__ + # ZMQ endpoints only. The MCP resident-server socket (openhcs/mcp/socket.py) + # is a different transport and already names its checkout; it is U4's. + for package in ("runtime", "agent", "core", "pyqt_gui"): + for source in (REPOSITORY_ROOT / "openhcs" / package).rglob("*.py"): + if source.name == "zmq_config.py": + continue + text = source.read_text(encoding="utf-8") + assert "ipc_socket_dir" not in text or "Path.home()" not in text, source + + +def test_the_harness_picks_no_free_port_for_an_execution_server(): + for source in (REPOSITORY_ROOT / "benchmark").glob("*.py"): + assert "DataControlPortPairAuthority" not in source.read_text( + encoding="utf-8" + ), source + + +def test_client_and_server_state_the_import_root_as_their_code_root(): + import_root = str(OpenHCSRuntimeImportAuthority.current().import_root) + + assert ZMQExecutionClient.code_root() == import_root + assert ZMQExecutionServer.code_root() == import_root + + class ScriptClient(ZMQExecutionClient): + """A subclass defined elsewhere is still this checkout's client.""" + + assert ScriptClient.code_root() == import_root + + +def test_each_checkout_has_its_own_socket_root(monkeypatch, tmp_path): + assert OPENHCS_ZMQ_CONFIG.socket_root() != ZMQConfig(app_name="openhcs").socket_root() + + monkeypatch.delenv(SOCKET_ROOT_ENV) + here = OPENHCS_ZMQ_CONFIG.socket_root() + assert here == checkout_socket_root() + + other_checkout = tmp_path / "checkout" + (other_checkout / "openhcs").mkdir(parents=True) + (other_checkout / "openhcs" / "__init__.py").write_text("", encoding="utf-8") + monkeypatch.setattr( + OpenHCSRuntimeImportAuthority, + "current", + classmethod(lambda cls: cls(other_checkout)), + ) + assert OPENHCS_ZMQ_CONFIG.socket_root() != here + + monkeypatch.setenv(SOCKET_ROOT_ENV, str(tmp_path)) + assert OPENHCS_ZMQ_CONFIG.socket_root() == tmp_path + + +def test_this_test_run_has_a_private_socket_root(): + root = OPENHCS_ZMQ_CONFIG.socket_root() + + assert root.parent == checkout_socket_root() + assert root.is_dir() + + +_CHILD_REPORT = """ +import json +from openhcs.runtime.zmq_execution_client import ZMQExecutionClient + +client = ZMQExecutionClient(port=65000, persistent=False) +print(json.dumps({ + "socket_root": str(client.config.socket_root()), + "code_root": client.launch().code_root, + "audience": client.launch().audience, + "control_url": client.endpoint.control_url(client.config), +})) +""" + + +def _report(environment: dict[str, str]) -> dict[str, str]: + completed = subprocess.run( + [sys.executable, "-c", _CHILD_REPORT], + env={**environment, "PYTHONPATH": str(REPOSITORY_ROOT)}, + capture_output=True, + text=True, + timeout=120, + ) + assert completed.returncode == 0, completed.stderr[-2000:] + return json.loads(completed.stdout.splitlines()[-1]) + + +def test_a_child_process_computes_its_parents_socket_root_and_launch(): + """A server or CLI started by this process looks where this process looks.""" + + client = ZMQExecutionClient(port=65000, persistent=False) + + assert _report(dict(os.environ)) == { + "socket_root": str(client.config.socket_root()), + "code_root": client.launch().code_root, + "audience": client.launch().audience, + "control_url": client.endpoint.control_url(client.config), + } + + +def test_a_private_socket_root_can_be_bound_whatever_tmpdir_is(tmp_path): + """The test run's sockets do not move with ``TMPDIR`` (a deep one cannot bind).""" + + deep_tmpdir = tmp_path / ("d" * 120) + deep_tmpdir.mkdir() + child = ( + "from openhcs.runtime.zmq_config import use_private_socket_root\n" + "use_private_socket_root()\n" + _CHILD_REPORT + ) + completed = subprocess.run( + [sys.executable, "-c", child], + env={ + **os.environ, + "PYTHONPATH": str(REPOSITORY_ROOT), + "TMPDIR": str(deep_tmpdir), + }, + capture_output=True, + text=True, + timeout=120, + ) + assert completed.returncode == 0, completed.stderr[-2000:] + report = json.loads(completed.stdout.splitlines()[-1]) + + assert Path(report["socket_root"]).parent == checkout_socket_root() + assert not Path(report["socket_root"]).exists() + require_bindable_address(report["control_url"]) + + +def test_a_client_refuses_a_socket_path_no_server_can_bind(monkeypatch, tmp_path): + """It raises before starting a server, and names the path and the limit.""" + + root = tmp_path / ("r" * 120) + monkeypatch.setenv(SOCKET_ROOT_ENV, str(root)) + client = ZMQExecutionClient(port=65000, persistent=False) + monkeypatch.setattr( + client, "_spawn_server_process", lambda: pytest.fail("a server was started") + ) + + with pytest.raises(SocketPathTooLongError) as refusal: + client.connect(timeout=5) + + assert str(root) in str(refusal.value) + assert str(zmq.IPC_PATH_MAX_LEN) in str(refusal.value) + assert SOCKET_ROOT_ENV in str(refusal.value) diff --git a/tests/unit/test_master_multi_plate_demo.py b/tests/unit/test_master_multi_plate_demo.py index a43a7366d..b1d0382c8 100644 --- a/tests/unit/test_master_multi_plate_demo.py +++ b/tests/unit/test_master_multi_plate_demo.py @@ -6,7 +6,10 @@ import pytest from polystore.streaming.identity import StreamProducerIdentity +from zmqruntime.transport import get_control_url, get_zmq_transport_url +from zmqruntime.transport_modes import TransportMode +from openhcs.agent.services.viewer_endpoint_discovery import viewer_ipc_directory from openhcs.core.artifacts.kinds import ImageArtifactType, MeasurementsArtifactType from openhcs.core.steps.function_step import FunctionStep from openhcs.processing.presets.demo_contribution import PipelineDemoContribution @@ -1354,5 +1357,15 @@ def test_master_viewer_endpoint_uses_the_openhcs_transport_owner(tmp_path): endpoint = master.McpMasterDemoOperations._endpoint(item) assert endpoint.config is master.OPENHCS_ZMQ_CONFIG - assert endpoint.data_url().endswith("/.openhcs/ipc/openhcs-zmq-6200.sock") - assert endpoint.control_url().endswith("/.openhcs/ipc/openhcs-zmq-7200.sock") + config = master.OPENHCS_ZMQ_CONFIG + socket_directory = config.socket_root() / config.ipc_socket_dir + assert viewer_ipc_directory() == socket_directory + assert endpoint.data_url() == f"ipc://{socket_directory}/openhcs-zmq-6200.sock" + assert endpoint.control_url() == f"ipc://{socket_directory}/openhcs-zmq-7200.sock" + # An execution endpoint on the same port resolves the same address. + assert endpoint.data_url() == get_zmq_transport_url( + 6200, mode=TransportMode.IPC, config=config + ) + assert endpoint.control_url() == get_control_url( + 6200, TransportMode.IPC, config=config + ) diff --git a/tests/unit/test_matched_cellprofiler_batch.py b/tests/unit/test_matched_cellprofiler_batch.py index c9c7a8bef..9949dfe78 100644 --- a/tests/unit/test_matched_cellprofiler_batch.py +++ b/tests/unit/test_matched_cellprofiler_batch.py @@ -361,9 +361,10 @@ def reject_endpoint(*args, **kwargs): monkeypatch.setattr(matched_batch, "_run_case", admitted_case) monkeypatch.setattr( - matched_batch.DataControlPortPairAuthority, "acquire", reject_endpoint + matched_batch, + "ZMQExecutionClient", + SimpleNamespace(with_private_server=reject_endpoint), ) - monkeypatch.setattr(matched_batch, "ZMQExecutionClient", reject_endpoint) output = tmp_path / "preflight" assert ( matched_batch.main( diff --git a/tests/unit/test_openhcs_adapter.py b/tests/unit/test_openhcs_adapter.py index 060e3b9de..5cd8bbf18 100644 --- a/tests/unit/test_openhcs_adapter.py +++ b/tests/unit/test_openhcs_adapter.py @@ -251,6 +251,10 @@ class FakeZMQExecutionClient: submitted: list[OpenHCSExecutionSubmission] = [] waits: list[str] = [] + @classmethod + def with_private_server(cls, *, progress_callback=None): + return cls(port=None, persistent=False, progress_callback=progress_callback) + def __init__(self, *, port, persistent, progress_callback): assert port is None self.progress_callback = progress_callback @@ -420,6 +424,10 @@ def test_benchmark_rejects_incompatible_execution_endpoint( tmp_path: Path, ) -> None: class IncompatibleZMQExecutionClient: + @classmethod + def with_private_server(cls, *, progress_callback=None): + return cls(port=None, persistent=False, progress_callback=progress_callback) + def __init__(self, *, port, persistent, progress_callback): self.connected_endpoint = SimpleNamespace( application=EndpointApplication("openhcs", "0.8.4"), diff --git a/tests/unit/test_openhcs_measured_run.py b/tests/unit/test_openhcs_measured_run.py index 3ae692b7a..de9ac01df 100644 --- a/tests/unit/test_openhcs_measured_run.py +++ b/tests/unit/test_openhcs_measured_run.py @@ -51,6 +51,10 @@ def test_measured_run_validates_an_ordinary_pipeline_document( clients = [] class FakeClient: + @classmethod + def with_private_server(cls, *, progress_callback=None): + return cls(port=None, persistent=False, progress_callback=progress_callback) + def __init__(self, *, port, persistent, progress_callback): assert port is None assert persistent is False @@ -615,6 +619,10 @@ def test_measured_run_refuses_to_reuse_an_unowned_server( selected_ports = [] class AttachedClient: + @classmethod + def with_private_server(cls, *, progress_callback=None): + return cls(port=None, persistent=False, progress_callback=progress_callback) + def __init__(self, **kwargs): selected_ports.append(kwargs["port"]) self.connected_endpoint = SimpleNamespace( @@ -634,11 +642,6 @@ def owned_server_process_is_alive(self): return None monkeypatch.setattr(measured_run, "ZMQExecutionClient", AttachedClient) - monkeypatch.setattr( - measured_run.DataControlPortPairAuthority, - "acquire", - lambda *_args, **_kwargs: SimpleNamespace(data_port=5555), - ) submission = OpenHCSExecutionSubmission( subject_id=tmp_path, pipeline_document=PipelineDocumentCodec.from_values( @@ -660,4 +663,4 @@ def owned_server_process_is_alive(self): timing_observer=measured_run._ZMQProgressTimingObserver(), require_owned_server=True, ) - assert selected_ports == [5555] + assert selected_ports == [None] diff --git a/tests/unit/test_well_throughput_scaling.py b/tests/unit/test_well_throughput_scaling.py index 5120cd0bc..0ff6d58a8 100644 --- a/tests/unit/test_well_throughput_scaling.py +++ b/tests/unit/test_well_throughput_scaling.py @@ -90,19 +90,12 @@ def test_reused_server_suite_keeps_one_client_and_distinct_resume_identity( monkeypatch.setattr( well_throughput_scaling, "load_comparison_cases", lambda _: cases ) - monkeypatch.setattr( - well_throughput_scaling, - "DataControlPortPairAuthority", - SimpleNamespace( - acquire=lambda *_args, **_kwargs: SimpleNamespace(data_port=7801) - ), - ) opened: list[object] = [] closed: list[object] = [] class FakeClient: def __init__(self, *, port, persistent, progress_callback): - assert (port, persistent) == (7801, False) + assert (port, persistent) == (None, False) self.progress_callback = progress_callback def __enter__(self): @@ -112,7 +105,13 @@ def __enter__(self): def __exit__(self, *_args): closed.append(self) - monkeypatch.setattr(well_throughput_scaling, "ZMQExecutionClient", FakeClient) + monkeypatch.setattr( + well_throughput_scaling, + "measured_execution_client", + lambda *, execution_port, progress_callback: FakeClient( + port=execution_port, persistent=False, progress_callback=progress_callback + ), + ) observed: list[tuple[str, object]] = [] def fake_run_case_well_throughput(**kwargs): diff --git a/tests/unit/test_zmq_execution_server_process.py b/tests/unit/test_zmq_execution_server_process.py index f0925df0b..132c39096 100644 --- a/tests/unit/test_zmq_execution_server_process.py +++ b/tests/unit/test_zmq_execution_server_process.py @@ -114,6 +114,11 @@ class _Server: def __init__(self, **_kwargs) -> None: events.append("construct") + def launched_by(self, launch): + # No --launch argument: a server started by hand owns itself. + assert launch.owner == ProcessIdentity.current() + return self + def start(self) -> None: events.append("start") @@ -149,6 +154,9 @@ class _Server: def __init__(self, **kwargs) -> None: captured.update(kwargs) + def launched_by(self, _launch): + return self + def prepare_runtime_capabilities(self, _status_callback) -> None: return None