add mixin for dra device behaviour

docs/explanations/connections.md

@@ -178,6 +178,39 @@ Each attempt closes the link and reopens it. `reconnect_attempts` consecutive
 failures is terminal until the process restarts; a clean connection restores the
 budget.
 
+Some failures are known never to recover, and retrying them is noise. A connection
+holds a `Recovery` policy that says which: a failed reconnect the policy calls
+terminal gives up at once instead of burning the rest of the budget, and the
+"Giving up" log line says why. The default policy calls nothing terminal. A device
+node injected by a Kubernetes DRA claim is the case that motivated it - the claim is
+made when the pod starts, so a node that has gone will not come back without a
+restart:
+
+```python
+from fastcs.connections import DRANode, SerialConnection
+
+
+class StageConnection(SerialConnection):
+    recovery = DRANode()
+```
+
+`DRANode` is also *fatal*: rather than stall with its dependents serving stale
+values, it asks the runner to shut the application down, so the orchestrator
+restarts the pod and the claim is re-established. A policy that is terminal but not
+fatal gives up and stalls, like a spent budget.
+
+The policy is held rather than inherited, so the transport and what to do when it
+fails are chosen separately. The same `DRANode()` serves a serial port, a socket or
+anything else, with no class per transport × policy, and it can be assigned to a
+single instance (`connection.recovery = DRANode()`) as well as set on a class. It
+is not a constructor argument, so it does not appear in the connection's
+configuration. A policy names the device it gave up on by the connection's `label` -
+the port, address or URL where the connection knows one.
+
+`reconnect_period` and `reconnect_attempts` stay on the connection rather than the
+policy: whether a failure is terminal is a fact about the device, while the period
+and the budget are what a site tunes.
+
 ### Dependencies
 
 A connection layered over others declares them, rather than having them derived from

docs/explanations/decisions/0021-connection-recovery-policy.md

@@ -0,0 +1,96 @@
+# 21. A connection holds its recovery policy rather than inheriting it
+
+Date: 2026-09-11
+
+## Status
+
+Accepted
+
+## Context
+
+A connection whose reconnect fails is retried every `reconnect_period` until
+`reconnect_attempts` consecutive failures, then gives up and stalls its dependents
+until the process restarts. For some failures that is the wrong shape. A device node
+injected by a Kubernetes DRA claim - a USB/IP serial port, say - is established when
+the pod starts; once it has gone it will not reappear in that pod, so every retry
+is noise and the connection then sits there, apparently healthy, until someone
+notices. The only fix is a pod restart.
+
+Commit `0f62b58` expressed this by inheritance. `Connection` gained two hooks,
+`is_terminal(exc)` and `unrecoverable_reason()`, and `fastcs.connections.dra` added a
+`DRADeviceMixin` that overrode them, requiring a driver to implement an abstract
+`_node_path` property:
+
+```python
+class DRASerialConnection(DRADeviceMixin, SerialConnection):
+    @property
+    def _node_path(self) -> str:
+        return self._settings.port
+```
+
+The runner never consulted either hook, so the mixin had no effect yet. Porting
+`fastcs-ximc` to it surfaced four problems, all of them about inheritance rather than
+the behaviour:
+
+1. **MRO order was a silent trap.** `class X(DRADeviceMixin, SerialConnection)`
+   worked; `class X(SerialConnection, DRADeviceMixin)` inherited
+   `Connection.is_terminal`, returned `False`, and the mixin did nothing - no error,
+   no warning, nothing in a diff to notice.
+2. **A class per transport × policy.** Serial needed `DRASerialConnection`; a claimed
+   IP or HTTP device needed another, and a second policy would multiply them again. A
+   claimed node behaves identically behind any transport, so this was duplication
+   with no content.
+3. **The policy could not change without changing the class.** Whether a node comes
+   from a DRA claim is a deployment fact, but inheritance fixed it at authoring time.
+4. **The contract was a private abstract hook.** `_node_path` was what a driver had
+   to implement, and it appeared in no public signature.
+
+## Decision
+
+Replace the mixin with a policy object the connection holds.
+
+- `fastcs.connections.recovery` defines `Recovery`, the default: it calls no failure
+  terminal. It has `is_terminal(exc)`, `reason(connection)` for the log line, and
+  `is_fatal`, which says whether a terminal failure should bring the application
+  down. `DRANode` is the first subclass: `FileNotFoundError` is terminal, and fatal.
+  Policies are stateless, so one instance can be shared.
+- `Connection.recovery` is a class attribute defaulting to `Recovery()`. A connection
+  class sets it, or it is assigned to one instance. It is deliberately not a
+  constructor argument: the launcher builds a connection's config schema from its
+  `__init__` signature, so an argument would appear in every connection's schema,
+  and choosing a policy in YAML would need a discriminated union.
+- `Connection.is_terminal` and `Connection.unrecoverable_reason` are removed rather
+  than kept as delegating wrappers, which would give two ways to say one thing with
+  undefined precedence when a subclass both overrode a method and held a policy.
+- `Connection.label` is a public property naming the device - the port, address or
+  base URL for the framework connections, and the class name otherwise. It replaces
+  `_node_path`.
+- The `ControllerRunner` consults the policy on every failed reconnect. A terminal
+  failure gives up immediately, without spending the rest of the retry budget; if
+  the policy is fatal, the runner also reports it through `fatal_error`, which
+  `FastCS.serve` already turns into a clean shutdown. The first `connect` at startup
+  is not affected, since a failure there already aborts startup.
+- `fastcs.connections.dra` is deleted.
+
+`reconnect_period` and `reconnect_attempts` stay on the connection. Whether a failure
+is terminal is a device fact that no deployment should be able to contradict; the
+period and the budget are what a site tunes, and are configurable per connection in
+`fastcs.yaml`. Folding them into the policy would mean either a site could overrule
+the device fact or the numbers stopped being configurable.
+
+## Consequences
+
+One policy serves any transport, and a connection is given one by assignment, so the
+four problems above go away: there is no base-class order to get wrong, no class per
+combination, a policy can be changed on an instance, and the contract a policy relies
+on (`label`) is public.
+
+A DRA-claimed device whose node disappears now gives up on the first failed reconnect
+and shuts the application down with a message saying why, so the orchestrator
+restarts the pod and the claim is re-established.
+
+The policy is not selectable from `fastcs.yaml`; the connection class carries the
+choice. That can be revisited if a deployment needs to change policy without
+changing `type:`. A later extension could let a policy own the retry *schedule*
+(for example a backoff computed from the connection's `reconnect_period`) while the
+connection keeps owning the numbers; that is a separate decision.

src/fastcs/connections/__init__.py

@@ -7,6 +7,8 @@
 from .ip_connection import IPConnection as IPConnection
 from .ip_connection import IPConnectionSettings as IPConnectionSettings
 from .ip_connection import StreamConnection as StreamConnection
+from .recovery import DRANode as DRANode
+from .recovery import Recovery as Recovery
 from .registry import Connections as Connections
 from .serial_connection import SerialConnection as SerialConnection
 from .serial_connection import SerialConnectionSettings as SerialConnectionSettings

src/fastcs/connections/connection.py

@@ -4,6 +4,8 @@
 from abc import ABC, abstractmethod
 from collections.abc import Sequence
 
+from fastcs.connections.recovery import Recovery
+
 DEFAULT_RECONNECT_PERIOD = 1.0
 """Seconds a connection waits between reconnect attempts, unless it says otherwise."""
 
@@ -68,6 +70,14 @@ async def get(self, path: str):
     reconnect_period: float = DEFAULT_RECONNECT_PERIOD
     reconnect_attempts: int = DEFAULT_RECONNECT_ATTEMPTS
 
+    recovery: Recovery = Recovery()
+    """What to do when this connection fails; assign a policy to change it.
+
+    A class attribute, not a constructor argument: a constructor argument would
+    appear in every connection's config schema. Policies are stateless, so the
+    default instance is shared.
+    """
+
     def __init__(
         self,
         depends_on: Connection | Sequence[Connection] | None = None,
@@ -139,5 +149,15 @@ async def wait_down(self) -> None:
         """Block until this connection is down. Returns immediately if it already is."""
         await self._down.wait()
 
+    @property
+    def label(self) -> str:
+        """What to call this connection's device in a failure message.
+
+        The device node or address where a connection knows one, and the class
+        name otherwise. Distinct from the role name the runner logs, which comes
+        from config rather than from the device.
+        """
+        return type(self).__name__
+
     def __repr__(self) -> str:
         return f"{type(self).__name__}(connected={self._connected})"

src/fastcs/connections/http_connection.py

@@ -55,6 +55,10 @@ def __init__(
 
         self.__client: AsyncClient | None = None
 
+    @property
+    def label(self) -> str:
+        return self._settings.base_url
+
     @property
     def _client(self) -> AsyncClient:
         if self.__client is None:

src/fastcs/connections/ip_connection.py

@@ -72,6 +72,10 @@ def __init__(self, settings: IPConnectionSettings | None = None, **kwargs) -> No
         self._settings = settings or IPConnectionSettings()
         self.__connection: StreamConnection | None = None
 
+    @property
+    def label(self) -> str:
+        return f"{self._settings.ip}:{self._settings.port}"
+
     @property
     def _connection(self) -> StreamConnection:
         if self.__connection is None:

src/fastcs/connections/recovery.py

@@ -0,0 +1,57 @@
+"""What to do about a connection failure (ADR 0021).
+
+A policy a connection holds, rather than a base class it inherits: the transport
+and what to do when it fails are chosen separately, so one policy serves any
+connection and a connection can be given any policy.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from fastcs.connections.connection import Connection
+
+
+class Recovery:
+    """Keep retrying. The default for every connection.
+
+    Stateless, so one instance can be shared between any number of connections.
+    """
+
+    is_fatal: bool = False
+    """Whether a terminal failure should bring the application down.
+
+    A connection that has given up stalls its dependents and serves stale values
+    forever. When only a restart can fix the cause, saying so and exiting is
+    better than sitting there looking healthy.
+    """
+
+    def is_terminal(self, exc: BaseException) -> bool:
+        """Whether a failed `Connection.connect` can never succeed in this process."""
+        return False
+
+    def reason(self, connection: Connection) -> str:
+        """Why it cannot recover, for the log line that ends the retry loop."""
+        return f"{connection.label} cannot recover from this failure."
+
+
+class DRANode(Recovery):
+    """A device node injected by a Kubernetes DRA claim.
+
+    The claim is established when the pod starts, so a node that has gone will
+    not reappear in it. Retrying cannot help and a pod restart can, so this is
+    both terminal and fatal.
+    """
+
+    is_fatal = True
+
+    def is_terminal(self, exc: BaseException) -> bool:
+        return isinstance(exc, FileNotFoundError)
+
+    def reason(self, connection: Connection) -> str:
+        return (
+            f"Device node {connection.label} has gone away. It comes from a "
+            "Kubernetes DRA claim and will not reappear in this pod. Restart "
+            "the pod to re-establish the claim."
+        )

src/fastcs/connections/serial_connection.py

@@ -37,6 +37,10 @@ def __init__(self, settings: SerialConnectionSettings, **kwargs) -> None:
         self._lock = asyncio.Lock()
         self.__stream: aioserial.AioSerial | None = None
 
+    @property
+    def label(self) -> str:
+        return self._settings.port
+
     async def connect(self) -> None:
         self.__stream = aioserial.AioSerial(
             port=self._settings.port, baudrate=self._settings.baud

src/fastcs/controllers/runner.py

@@ -109,10 +109,9 @@ def __init__(
         here instead. `FastCS` awaits it and shuts down; an embedder can do the same,
         and read `fatal_reason` for what happened.
 
-        Nothing in the framework sets this today: its one producer was the
-        introspection mismatch on reconnect, which went with introspection itself.
-        The channel is kept because the problem it solves - a background task that
-        cannot raise - has not gone anywhere.
+        Set by a reconnect that fails terminally under a fatal `Recovery` policy -
+        a DRA device node that has gone away, say - since only a restart can fix
+        that.
         """
 
         self.fatal_reason: BaseException | None = None
@@ -492,9 +491,13 @@ async def _attempt(self, connection: Connection) -> None:
         try:
             await connection.close()  # tolerate an already-closed link
             await connection.connect()
-        except Exception:
+        except Exception as exc:
             logger.exception("Reconnect failed", connection=self._name_of(connection))
-            if state.attempts >= connection.reconnect_attempts:
+            recovery = connection.recovery
+            # A failure the policy knows cannot recover gives up at once, rather
+            # than spending the rest of the budget on retries that cannot succeed.
+            terminal = recovery.is_terminal(exc)
+            if terminal or state.attempts >= connection.reconnect_attempts:
                 # Terminal until the process restarts. Setting the event releases
                 # anything waiting on this connection, so dependents stall loudly
                 # instead of hanging silently.
@@ -503,11 +506,16 @@ async def _attempt(self, connection: Connection) -> None:
                     "Giving up",
                     connection=self._name_of(connection),
                     attempts=state.attempts,
+                    reason=recovery.reason(connection) if terminal else None,
                     blocks=[
                         self._name_of(dependent)
                         for dependent in self._dependents_of(connection)
                     ],
                 )
+                if terminal and recovery.is_fatal:
+                    # Only a restart can fix it, so ask for one rather than sit
+                    # there looking healthy while serving stale values.
+                    self.fail(exc)
             return
 
         connection._set_connected()  # noqa: SLF001