Fix reply correlation and failure reporting

docs/api/device.md

@@ -3,6 +3,7 @@
 ---
 
 ::: harp.device.client.Device
+::: harp.device.client.DeviceError
 ::: harp.device.client.HarpFramer
 ::: harp.device.client.ITransport
 ::: harp.device.client.TransportError

src/packages/harp-device/README.md

@@ -14,6 +14,10 @@ who = device.read(core.WhoAmI).payload          # -> np.uint16
 device.write(core.OperationControl, payload)   # write a register
 ```
 
+## When a request fails
+
+An error reply raises `DeviceError`, which keeps the reply as `reply` so the frame sent by the device stays available for inspection. Pass `raise_on_error=False` to the constructor to receive such a reply as an ordinary return value instead. A transport failure raises `TransportError`, and every later request reports the same failure rather than waiting for a reply that cannot arrive. A device that never answers raises `TimeoutError` after `REPLY_TIMEOUT`, which is also what happens when `close` is called during a request.
+
 ## Extending for a specific device
 
 A device is described by a module. Downstream, often generated, packages record the device identity as `WHO_AM_I`, declare the register classes at module level, and expand the core `REGISTER_MAP` beside them:

src/packages/harp-device/src/harp/device/client/__init__.py

@@ -1,11 +1,12 @@
 """Talking to a Harp device: the device itself, its transport and the framer."""
 
-from ._device import Device, EventHandler, Subscription
+from ._device import Device, DeviceError, EventHandler, Subscription
 from ._framer import HarpFramer
 from ._transport import ITransport, TransportError
 
 __all__ = [
     "Device",
+    "DeviceError",
     "EventHandler",
     "Subscription",
     "HarpFramer",

src/packages/harp-device/src/harp/device/client/_device.py

@@ -31,13 +31,33 @@
 _DEFAULT_MESSAGE_TYPES: frozenset[MessageType] = frozenset({MessageType.Event})
 """Default filter for :meth:`Device.subscribe`: unsolicited events only."""
 
+_Waiter = queue.SimpleQueue["HarpMessage | Exception"]
+"""The received reply or failure that ends the wait of a blocked request."""
+
 
 def _normalize_message_types(message_types: MessageTypeFilter) -> frozenset[MessageType]:
     if isinstance(message_types, MessageType):
         return frozenset({message_types})
     return frozenset(message_types)
 
 
+class DeviceError(Exception):
+    """Raised when the device replies to a request with the error flag set.
+
+    The reply is kept as :attr:`reply`, so the frame sent by the device stays available
+    for inspection. Construct the device with ``raise_on_error=False`` to receive such
+    a reply as an ordinary return value instead.
+    """
+
+    def __init__(self, reply: HarpMessage) -> None:
+        super().__init__(
+            f"Device returned error for register address {reply.address} "
+            f"(0x{reply.address:02x}). Payload: {reply.payload_bytes.hex()}"
+        )
+        self.reply = reply
+        """The error reply, as received."""
+
+
 class Subscription:
     """Handle returned by :meth:`Device.subscribe`. Cancel with
     :meth:`unsubscribe`, or use as a context manager to auto-cancel on exit."""
@@ -85,6 +105,13 @@ class Device(Generic[M]):
     Omitting ``device_module`` skips that check. The module is not otherwise
     consulted: registers reach :meth:`read`, :meth:`write` and :meth:`subscribe`
     as arguments either way, and only a subscribed register is parsed on arrival.
+
+    A request fails in one of three ways. An error reply raises :class:`DeviceError`
+    carrying the reply, unless ``raise_on_error=False``. A transport failure raises
+    :class:`~harp.device.client.TransportError`, and every later request reports the
+    same failure rather than waiting for a reply that cannot arrive. A device that
+    never answers raises :class:`TimeoutError` after ``REPLY_TIMEOUT``, which is also
+    what happens when :meth:`close` is called during a request.
     """
 
     REPLY_TIMEOUT: ClassVar[float] = 5.0  # seconds
@@ -114,8 +141,9 @@ def __init__(
         self._device_module = device_module
         self.raise_on_error = raise_on_error
         self._framer = HarpFramer()
-        self._pending: dict[int, queue.SimpleQueue] = {}
+        self._pending: dict[tuple[int, MessageType], list[_Waiter]] = {}
         self._pending_lock = threading.Lock()
+        self._fault: Exception | None = None
         self._running = False
         self._thread: threading.Thread | None = None
 
@@ -207,7 +235,7 @@ def read(
         # Note: ty can't correctly infer the return type, and this is a known issue:
         # https://github.com/astral-sh/ty/issues/623
         frame = register.format(message_type=MessageType.Read, timestamp=timestamp, port=port)
-        msg = self._request(register.address, frame)
+        msg = self._request(register.address, MessageType.Read, frame)
         return msg.decode(register)
 
     def write(
@@ -221,7 +249,7 @@ def write(
         frame = register.format(
             value, message_type=MessageType.Write, timestamp=timestamp, port=port
         )
-        msg = self._request(register.address, frame)
+        msg = self._request(register.address, MessageType.Write, frame)
         return msg.decode(register)
 
     # ------------------------------------------------------------------
@@ -337,9 +365,10 @@ def _read_loop(self) -> None:
         while self._running:
             try:
                 chunk = self._transport.read()
-            except TransportError:
+            except TransportError as exc:
                 if self._running:
-                    raise
+                    _logger.exception("The transport failed, so the device stopped reading")
+                    self._fault_pending(exc)
                 break  # expected while shutting down
             if not chunk:
                 continue
@@ -348,37 +377,54 @@ def _read_loop(self) -> None:
             for msg in self._framer.frames():
                 self._dispatch(msg)
 
+    def _fault_pending(self, exc: Exception) -> None:
+        """Report ``exc`` to every waiting request, and record it so a later request
+        reports it at once instead of waiting for a reply that cannot arrive."""
+        with self._pending_lock:
+            self._fault = exc
+            waiting = [q for waiters in self._pending.values() for q in waiters]
+            self._pending.clear()
+        for q in waiting:
+            q.put(exc)
+
     def _dispatch(self, msg: HarpMessage) -> None:
         # Fast path: correlate replies to a pending synchronous request. This is
         # O(1) and non-blocking, so it should never stall behind a slow subscriber.
         # Events are unsolicited and never correlate to requests
         if msg.message_type != MessageType.Event:
             with self._pending_lock:
-                q = self._pending.get(msg.address)
-            if q is not None:
+                waiting = list(self._pending.get((msg.address, msg.message_type), ()))
+            for q in waiting:
                 q.put(msg)
 
         self._event_queue.put(msg)
 
-    def _request(self, address: int, frame: bytes) -> HarpMessage:
-        q: queue.SimpleQueue = queue.SimpleQueue()
+    def _request(self, address: int, message_type: MessageType, frame: bytes) -> HarpMessage:
+        key = (address, message_type)
+        q: _Waiter = queue.SimpleQueue()
         with self._pending_lock:
-            self._pending[address] = q
+            if self._fault is not None:
+                raise self._fault
+            self._pending.setdefault(key, []).append(q)
         try:
             self._transport.write(frame)
             try:
-                msg = q.get(timeout=self.REPLY_TIMEOUT)
-                if msg.has_error and self.raise_on_error:
-                    raise RuntimeError(
-                        f"Device returned error for register address {address} "
-                        f"(0x{address:02x}). Payload: {msg.payload_bytes.hex()}"
-                    )
-                return msg
+                reply = q.get(timeout=self.REPLY_TIMEOUT)
             except queue.Empty as exc:
                 raise TimeoutError(
                     f"No reply from device for register address {address} "
                     f"within {self.REPLY_TIMEOUT}s"
                 ) from exc
+            if isinstance(reply, Exception):
+                raise reply
+            if reply.has_error and self.raise_on_error:
+                raise DeviceError(reply)
+            return reply
         finally:
             with self._pending_lock:
-                self._pending.pop(address, None)
+                waiters = self._pending.get(key)
+                if waiters is not None:
+                    if q in waiters:
+                        waiters.remove(q)
+                    if not waiters:
+                        del self._pending[key]

tests/device/test_device.py

@@ -1,6 +1,19 @@
+import queue
+import threading
 import types
+from collections.abc import Callable, Iterable
+from concurrent.futures import ThreadPoolExecutor
 
-from harp.device.client import Device
+import numpy as np
+import pytest
+
+from harp.device import core
+from harp.device.client import Device, DeviceError, TransportError
+from harp.protocol import MessageType
+from tests.fixtures import make_frame_from_raw
+
+_U16 = 0x02
+"""Payload-type byte of a U16 payload, as the size nibble alone."""
 
 
 class _NullTransport:
@@ -14,6 +27,45 @@ def read(self) -> bytes:
         return b""
 
 
+class _ScriptedTransport:
+    """A transport replying with whatever ``on_write`` returns for each request.
+
+    Frames are queued from inside ``write``, which is reached only once the request is
+    registered, so a reply cannot be dispatched before there is a waiter to receive it.
+    Setting ``failing`` makes the next read fail, as a removed port would.
+    """
+
+    def __init__(self) -> None:
+        self.writes: list[bytes] = []
+        self.failing = False
+        self.on_write: Callable[[bytes], Iterable[bytes]] | None = None
+        self._inbox: queue.SimpleQueue[bytes] = queue.SimpleQueue()
+
+    def open(self) -> None: ...
+
+    def close(self) -> None: ...
+
+    def write(self, data: bytes) -> None:
+        self.writes.append(data)
+        if self.on_write is not None:
+            for frame in self.on_write(data):
+                self._inbox.put(frame)
+
+    def read(self) -> bytes:
+        if self.failing:
+            raise TransportError("simulated transport failure")
+        try:
+            return self._inbox.get(timeout=0.01)
+        except queue.Empty:
+            return b""
+
+
+class _ShortTimeoutDevice(Device[None]):
+    """A device that gives up on a reply quickly, so a test never waits five seconds."""
+
+    REPLY_TIMEOUT = 0.5
+
+
 def _module(name: str, **attrs: object) -> types.ModuleType:
     mod = types.ModuleType(name)
     mod.DEVICE_NAME = name
@@ -33,3 +85,94 @@ def test_module_is_returned_by_the_property():
     module = _module("Behavior", WHO_AM_I=0, REGISTER_MAP={})
     assert Device(_NullTransport(), module).module is module
     assert Device(_NullTransport()).module is None
+
+
+def test_write_reply_does_not_satisfy_read():
+    # A reply carries the message type of its request, so a write reply on the same
+    # register must not answer a pending read.
+    transport = _ScriptedTransport()
+    transport.on_write = lambda _: (
+        core.WhoAmI.format(np.uint16(7), message_type=MessageType.Write),
+        core.WhoAmI.format(np.uint16(9), message_type=MessageType.Read),
+    )
+    with _ShortTimeoutDevice(transport) as device:
+        assert int(device.read(core.WhoAmI).payload) == 9
+
+
+def test_event_does_not_satisfy_read():
+    # Events are unsolicited, so one for the same register must not answer a read.
+    transport = _ScriptedTransport()
+    transport.on_write = lambda _: (
+        core.WhoAmI.format(np.uint16(7), message_type=MessageType.Event),
+        core.WhoAmI.format(np.uint16(9), message_type=MessageType.Read),
+    )
+    with _ShortTimeoutDevice(transport) as device:
+        assert int(device.read(core.WhoAmI).payload) == 9
+
+
+def test_concurrent_reads_share_one_reply():
+    # The wire carries no request identifier, so two reads in flight on one register
+    # cannot be told apart in the reply. Both are answered by it rather than one
+    # replacing the waiter of the other.
+    transport = _ScriptedTransport()
+    both_in_flight = threading.Barrier(2)
+
+    def reply_once(_: bytes) -> Iterable[bytes]:
+        first = both_in_flight.wait(timeout=2) == 0
+        return () if first else (core.WhoAmI.format(np.uint16(9), message_type=MessageType.Read),)
+
+    transport.on_write = reply_once
+    with _ShortTimeoutDevice(transport) as device, ThreadPoolExecutor(2) as pool:
+        replies = [pool.submit(device.read, core.WhoAmI) for _ in range(2)]
+        assert [int(reply.result(timeout=2).payload) for reply in replies] == [9, 9]
+
+
+def test_transport_failure_faults_pending_read():
+    transport = _ScriptedTransport()
+
+    def fail(_: bytes) -> Iterable[bytes]:
+        transport.failing = True
+        return ()
+
+    transport.on_write = fail
+    with _ShortTimeoutDevice(transport) as device:
+        with pytest.raises(TransportError):
+            device.read(core.WhoAmI)
+
+
+def test_read_after_transport_failure_raises_transport_error():
+    # A failure that has already stopped the reader is reported to the next request
+    # rather than leaving it to time out on a reply that cannot arrive.
+    transport = _ScriptedTransport()
+
+    def fail(_: bytes) -> Iterable[bytes]:
+        transport.failing = True
+        return ()
+
+    transport.on_write = fail
+    with _ShortTimeoutDevice(transport) as device:
+        with pytest.raises(TransportError):
+            device.read(core.WhoAmI)
+        with pytest.raises(TransportError):
+            device.read(core.WhoAmI)
+    assert len(transport.writes) == 1  # the second request never reached the transport
+
+
+def test_error_reply_raises_device_error():
+    frame = make_frame_from_raw(MessageType.Read | 0x08, 0, 255, _U16, b"\x07\x00")
+    transport = _ScriptedTransport()
+    transport.on_write = lambda _: (frame,)
+    with _ShortTimeoutDevice(transport) as device:
+        with pytest.raises(DeviceError) as error:
+            device.read(core.WhoAmI)
+    assert error.value.reply.bytes == frame  # the frame is kept, not formatted away
+
+
+def test_error_reply_returned_when_not_raising():
+    frame = make_frame_from_raw(MessageType.Read | 0x08, 0, 255, _U16, b"\x07\x00")
+    transport = _ScriptedTransport()
+    transport.on_write = lambda _: (frame,)
+    with _ShortTimeoutDevice(transport, raise_on_error=False) as device:
+        reply = device.read(core.WhoAmI)
+    assert reply.has_error
+    assert int(reply.payload) == 7