diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..d1ef747 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,27 @@ +# Changelog + +## 0.6.5 + +Requires `python-introspect>=0.3.0,<0.4` (was `<0.3`). + +Wire form: +- No message declared by zmqruntime changes on the wire: none of them has a field + whose declared type admits several classes, so a 0.6.4 peer and a 0.6.5 peer + exchange the same bytes for zmqruntime's own messages. +- An application message does change if it declares such a field (a union of two + or more dataclasses, an abstract dataclass, or an `AutoRegisterMeta` class): the + value is sent as `{tag: fields}` and read by its tag. There is no reader for the + untagged form. zmqruntime has no protocol version of its own; a client refuses + a server of another application version only when the application declares an + `EndpointApplication(identifier, version)` and requires it, so applications + whose messages changed form must raise that version. +- A field declared as one concrete dataclass carries that class only. A subclass + instance with more fields is refused when decoded. + +Fixed: +- A control request encodes its body by the declared field types, the same way it + is decoded. Before, each body value was encoded on its own, so under + python-introspect 0.3 a request field admitting several classes would have + been sent untagged and then refused by the receiver. +- `ViewerInstance.visualizer` resolves at run time (`VisualizerProcessManager` + was imported only for type checkers). diff --git a/pyproject.toml b/pyproject.toml index 9cb6265..a8808b8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "zmqruntime" -version = "0.6.4" +version = "0.6.5" description = "Generic ZMQ-based distributed execution framework" authors = [{name = "Tristan Simas", email = "tristan.simas@mail.mcgill.ca"}] license = {text = "MIT"} @@ -31,7 +31,7 @@ dependencies = [ "portalocker>=2.8.2", "pyzmq>=22.0", "psutil>=5.9", - "python-introspect>=0.2.5,<0.3", + "python-introspect>=0.3.0,<0.4", "typing-extensions>=4.0.0", ] diff --git a/src/zmqruntime/messages.py b/src/zmqruntime/messages.py index 705250f..ecc13e9 100644 --- a/src/zmqruntime/messages.py +++ b/src/zmqruntime/messages.py @@ -469,9 +469,10 @@ def from_wire_payload(cls, wire_payload: bytes) -> Self: return cls.from_dict(payload) def to_dict(self) -> dict[str, object]: + declared = to_jsonable(replace(self, operation_deadline=None)) payload = { - MessageFields.TYPE: self.message_type.value, - **{name: to_jsonable(getattr(self, name)) for name in self.body_fields()}, + MessageFields.TYPE: declared["message_type"], + **{name: declared[name] for name in self.body_fields()}, } return (payload if self.operation_deadline is None else self.with_observation_budget(payload, self.operation_deadline)) diff --git a/src/zmqruntime/viewer_state.py b/src/zmqruntime/viewer_state.py index fa56a7f..72fa88e 100644 --- a/src/zmqruntime/viewer_state.py +++ b/src/zmqruntime/viewer_state.py @@ -16,13 +16,11 @@ from abc import ABC, abstractmethod from dataclasses import dataclass, field from enum import Enum, auto -from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple +from typing import Any, Callable, Dict, List, Optional, Tuple +from zmqruntime.streaming.process_manager import VisualizerProcessManager from zmqruntime.subscription import CallbackSubscription, SubscriptionABC -if TYPE_CHECKING: - from zmqruntime.streaming.process_manager import VisualizerProcessManager - logger = logging.getLogger(__name__) @@ -42,7 +40,7 @@ class ViewerInstance: viewer_type: str port: int - visualizer: "VisualizerProcessManager" + visualizer: VisualizerProcessManager state: ViewerState = ViewerState.LAUNCHING created_at: float = field(default_factory=time.time) last_used: float = field(default_factory=time.time) @@ -59,21 +57,21 @@ def is_healthy(self) -> bool: return self.visualizer.is_running -VisualizerFactory = Callable[[], "VisualizerProcessManager"] +VisualizerFactory = Callable[[], VisualizerProcessManager] class ViewerReuseAdmissionABC(ABC): """Caller-owned constraints admitted inside atomic viewer acquisition.""" @abstractmethod - def require_reusable(self, visualizer: "VisualizerProcessManager") -> None: + def require_reusable(self, visualizer: VisualizerProcessManager) -> None: """Raise without changing the existing instance when reuse is forbidden.""" class HealthyViewerReuseAdmission(ViewerReuseAdmissionABC): """Generic callers require health but declare no application launch settings.""" - def require_reusable(self, visualizer: "VisualizerProcessManager") -> None: + def require_reusable(self, visualizer: VisualizerProcessManager) -> None: pass @@ -141,7 +139,7 @@ def get_or_create_viewer( ready_timeout: float = 30.0, *, reuse_admission: ViewerReuseAdmissionABC = HealthyViewerReuseAdmission(), - ) -> Tuple["VisualizerProcessManager", bool]: + ) -> Tuple[VisualizerProcessManager, bool]: """ Atomically get existing viewer or create a new one. @@ -294,7 +292,7 @@ def _remove_viewer_locked(self, key: Tuple[str, int]) -> None: except Exception as e: logger.warning("Error stopping viewer during removal: %s", e) - def get_viewer(self, viewer_type: str, port: int) -> Optional["VisualizerProcessManager"]: + def get_viewer(self, viewer_type: str, port: int) -> Optional[VisualizerProcessManager]: """Get existing viewer if healthy.""" key = (viewer_type, port) with self._lock: @@ -509,7 +507,7 @@ def get_or_create_viewer( ready_timeout: float = 30.0, *, reuse_admission: ViewerReuseAdmissionABC = HealthyViewerReuseAdmission(), -) -> Tuple["VisualizerProcessManager", bool]: +) -> Tuple[VisualizerProcessManager, bool]: """ Convenience function for atomic viewer get-or-create. diff --git a/tests/test_messages.py b/tests/test_messages.py index bf5033a..9e46cd1 100644 --- a/tests/test_messages.py +++ b/tests/test_messages.py @@ -313,3 +313,44 @@ def test_task_progress_carries_the_application_phase_vocabulary() -> None: assert wire["phase"] == wire["status"] == "compiling" assert TaskProgress.from_dict(wire).phase == "compiling" + + +@dataclass(frozen=True) +class _ByName: + name: str + + +@dataclass(frozen=True) +class _ByAlias: + name: str + + +@dataclass(frozen=True, slots=True) +class _SelectRequest(ControlRequestHeader): + selector: _ByName | _ByAlias + + +def test_request_body_field_admitting_several_classes_is_tagged_on_the_wire() -> None: + request = _SelectRequest(ControlMessageType.STATUS, _ByAlias("a")) + + payload = request.to_dict() + + assert payload == {"type": "status", "selector": {"_ByAlias": {"name": "a"}}} + assert _SelectRequest.from_dict(payload) == request + assert _SelectRequest.from_wire_payload(request.to_wire_payload()) == request + with pytest.raises(ValueError, match="_ByName"): + _SelectRequest.from_dict({"type": "status", "selector": {"Unknown": {"name": "a"}}}) + + +@dataclass(frozen=True) +class _NamedProcessIdentity(ProcessIdentity): + name: str = "" + + +def test_field_declared_as_one_class_refuses_a_subclass_with_more_fields() -> None: + from zmqruntime import messages + + launch = messages.EndpointLaunch("/code", _NamedProcessIdentity(1, 2.0, "n"), "any_session") + + with pytest.raises(ValueError, match="name"): + messages.EndpointLaunch.from_dict(launch.to_dict())