Restart safety: transient credential checks are 1013/503, not a refusal

- A device credential that cannot be checked now (device store not running,
  nonces unreadable) raises DeviceUnavailable: 503 over HTTP, and the
  /flows/ws handshake and the tunnel stream accept and close with 1013.
  1008 (a 403 before accepting) stays for a revoked or invalid one
- Device files are written through a temporary file, so a read mid-save no
  longer looks like a deleted device
- The first report turns Touch friendly on only for a device paired since
  pairings are noted (DeviceState.paired_at), so devices paired before keep
  the switch they were given
- NodeDef reads `device`/`device_policy` from older clients as the worker
  label, for a release
- Tests: transient vs revoked socket, the touch default, the old worker keys,
  and a restart with a device node and a node_health rule sending no push

backend/fluksio/api/deps.py

@@ -19,7 +19,7 @@ from fluksio.flow import devices
 from fluksio.flow.artifacts import is_reference
 from fluksio.flow.controller import FlowController
 from fluksio.flow.dashboards import DashboardDef, DashboardStore
-from fluksio.flow.devices import DeviceDef, DeviceStore, presence
+from fluksio.flow.devices import DeviceDef, DeviceStore, DeviceUnavailable, presence
 from fluksio.flow.workers import PythonWorkerPool
 from fluksio.models import OAuthClient, TokenPayload, User, WorkerToken
 
@@ -59,12 +59,15 @@ def _device_for(payload: dict[str, Any], app: Any) -> DeviceDef:
 
     The claim is still called ``panel``: tokens in circulation say it, and
     nothing pairs again for a rename.
+
+    What cannot be checked right now raises ``DeviceUnavailable`` rather than
+    refusing: a device told its credential is refused forgets it.
     """
-    store: DeviceStore | None = (
-        getattr(app.state, "device_store", None) if app else None
-    )
+    if app is None:
+        raise InvalidTokenError("Nothing to check a device credential against")
+    store: DeviceStore | None = getattr(app.state, "device_store", None)
     if store is None:
-        raise InvalidTokenError("The device store is not running")
+        raise DeviceUnavailable("The device store is not running")
     device = store.find(str(payload.get("panel", "")))
     if device is None:
         raise InvalidTokenError("This device no longer exists")
@@ -72,7 +75,7 @@ def _device_for(payload: dict[str, Any], app: Any) -> DeviceDef:
         try:
             nonce = store.nonce(device.id)
         except (OSError, ValueError) as exc:
-            raise InvalidTokenError(f"The device nonces are unreadable: {exc}")
+            raise DeviceUnavailable(f"The device nonces are unreadable: {exc}")
         if int(payload.get("pnc") or 0) != nonce:
             raise InvalidTokenError("This device was paired again")
     presence.seen(device.id)

backend/fluksio/api/routes/devices.py

@@ -625,7 +625,8 @@ async def put_report(device_id: str, request: Request, store: DeviceStoreDep) ->
 
     The first report of a touch screen makes the device touch friendly, as
     the switch would have been set by hand. Later ones leave the setting to
-    whoever changed it since.
+    whoever changed it since, and so does the first report of a device paired
+    before pairings were noted: its switch was set by hand already.
     """
     device: DeviceDef | None = getattr(request.state, "device", None)
     if device is None or device.id != device_id:
@@ -650,7 +651,13 @@ async def put_report(device_id: str, request: Request, store: DeviceStoreDep) ->
             ),
         )
     first = await run_in_threadpool(store.report, device_id, report)
-    if first and report.display and report.display.touch and not device.touch:
+    if (
+        first
+        and report.display
+        and report.display.touch
+        and not device.touch
+        and (await run_in_threadpool(store.state, device_id)).paired_at
+    ):
         with suppress(StaleVersion, DeviceNotFound):
             await run_in_threadpool(
                 store.write,
@@ -724,6 +731,7 @@ def _adopt(
         )
     entry.device_id = device_id
     _pending.save(code, entry)
+    store.paired(device_id)
     return f"Paired {entry.name or entry.device} with {device_id}"
 
 

backend/fluksio/api/routes/flows.py

@@ -1,6 +1,7 @@
 """The flow API. Everything the editor can do is available here first."""
 
 import asyncio
+import logging
 import os
 import time
 from pathlib import Path
@@ -68,6 +69,8 @@ from fluksio.flow.store import (
 )
 from fluksio.models import Flavor, Message, Run
 
+logger = logging.getLogger(__name__)
+
 router = APIRouter(
     prefix="/flows", tags=["flows"], dependencies=[Depends(get_current_user)]
 )
@@ -1221,15 +1224,25 @@ async def flow_events(websocket: WebSocket, ticket: str = "", token: str = "") -
     # every other socket and every request is sharing.
     def _authenticate() -> Any:
         with Session(engine) as session:
-            return user_from_token(session, token, app=websocket.app)
+            user = user_from_token(session, token, app=websocket.app)
+        only = panel_scope(token, websocket.app) if user is not None else None
+        # A device is online for as long as this socket is open.
+        device = device_of(token, websocket.app) if only is not None else ""
+        return user, only, device
 
-    user = await run_in_threadpool(_authenticate)
+    try:
+        user, only, device = await run_in_threadpool(_authenticate)
+    except Exception as exc:
+        # Not a refusal: the database or the device store could not say. A
+        # device takes 1008, or a 403 before accepting, as its credential
+        # being gone and checks it, so this one is told to try again later.
+        logger.warning("Could not check a socket's credential: %s", exc)
+        await websocket.accept()
+        await websocket.close(code=1013)
+        return
     if user is None:
         await websocket.close(code=1008)
         return
-    only = panel_scope(token, websocket.app)
-    # A device is online for as long as this socket is open.
-    device = device_of(token, websocket.app) if only is not None else ""
     # Nothing until a client says it is drawing something: bytes are what this
     # socket cannot afford to send speculatively.
     wanted: set[str] = set()

backend/fluksio/cloud/connector.py

@@ -663,15 +663,21 @@ class CloudConnector:
         if not path.startswith("/api/v1/") or not path.endswith("/flows/ws"):
             await socket.send(_dump({"op": "ws_close", "id": stream_id, "code": 1008}))
             return
-        with Session(engine) as session:
-            if user_from_token(session, token, app=self._app) is None:
-                await socket.send(
-                    _dump({"op": "ws_close", "id": stream_id, "code": 1008})
-                )
-                return
-        only = panel_scope(token, self._app)
-        # A remote device is online for as long as this stream is open.
-        device = device_of(token, self._app) if only is not None else ""
+        try:
+            with Session(engine) as session:
+                user = user_from_token(session, token, app=self._app)
+            only = panel_scope(token, self._app) if user is not None else None
+            # A remote device is online for as long as this stream is open.
+            device = device_of(token, self._app) if only is not None else ""
+        except Exception as exc:
+            # Could not tell, which is not a refusal: 1008 is what makes a
+            # device check whether it is still paired.
+            logger.warning("Could not check a stream's credential: %s", exc)
+            await socket.send(_dump({"op": "ws_close", "id": stream_id, "code": 1013}))
+            return
+        if user is None:
+            await socket.send(_dump({"op": "ws_close", "id": stream_id, "code": 1008}))
+            return
 
         await socket.send(_dump({"op": "ws_open_ok", "id": stream_id}))
 

backend/fluksio/flow/devices.py

@@ -159,6 +159,9 @@ class DeviceState(BaseModel):
     #: What the device says it has, written by the device itself.
     report: dict[str, Any] | None = None
     reported_at: datetime | None = None
+    #: When a pairing here last minted the device a credential. None for one
+    #: paired before this was kept, whose settings were all chosen by hand.
+    paired_at: datetime | None = None
 
 
 _STATES = TypeAdapter(dict[str, DeviceState])
@@ -231,6 +234,15 @@ class DeviceExists(ValueError):
         self.id = device_id
 
 
+class DeviceUnavailable(RuntimeError):
+    """Whether a device credential stands cannot be told right now.
+
+    Not a refusal: a device told its credential is refused pairs again, so
+    a store that is not running or a nonce file that does not read answers
+    "try again later" instead.
+    """
+
+
 class DeviceStore:
     """Device definitions in the flow repository, their nonces beside it.
 
@@ -260,7 +272,11 @@ class DeviceStore:
     def _write(self, device: DeviceDef) -> None:
         path = self._file(device.id)
         path.parent.mkdir(parents=True, exist_ok=True)
-        path.write_text(device.model_dump_json(indent=2) + "\n")
+        # Through a temporary file: a credential checked against a half
+        # written one would read as a deleted device and be refused.
+        tmp = path.with_name(path.name + ".tmp")
+        tmp.write_text(device.model_dump_json(indent=2) + "\n")
+        os.replace(tmp, path)
 
     # -- what the published flows wire ------------------------------------
 
@@ -433,6 +449,13 @@ class DeviceStore:
             self._write_states(states)
             return state.nonce
 
+    def paired(self, device_id: str) -> None:
+        """Note that a pairing here just minted this device a credential."""
+        with self._state_lock:
+            states = self._read_states()
+            states.setdefault(device_id, DeviceState()).paired_at = datetime.now(UTC)
+            self._write_states(states)
+
     def report(self, device_id: str, report: DeviceReport) -> bool:
         """Keep what the device says it has, in place of what it said before.
 

backend/fluksio/flow/schemas.py

@@ -226,6 +226,27 @@ class NodeDef(BaseModel):
         ),
     )
 
+    @model_validator(mode="before")
+    @classmethod
+    def _old_worker_keys(cls, data: Any) -> Any:
+        """Read ``device`` and ``device_policy`` as the worker label they were.
+
+        What a ``fluksio sync`` or SDK from before the rename to ``worker``
+        sends, which would otherwise be dropped without a word. A new key
+        sent beside an old one wins.
+
+        ponytail: drop after a release, once clients have caught up.
+        """
+        if not isinstance(data, dict) or not (
+            "device" in data or "device_policy" in data
+        ):
+            return data
+        data = dict(data)
+        for old, new in (("device", "worker"), ("device_policy", "worker_policy")):
+            if old in data:
+                data.setdefault(new, data.pop(old))
+        return data
+
     @field_validator("id")
     @classmethod
     def _check_id(cls, value: str) -> str:

backend/fluksio/main.py

@@ -34,7 +34,7 @@ from fluksio.flow.alerts import AlertManager
 from fluksio.flow.artifacts import ArtifactStore, VolatileStore
 from fluksio.flow.controller import FlowController, RebuildBusy
 from fluksio.flow.dashboards import DashboardStore
-from fluksio.flow.devices import DeviceStore
+from fluksio.flow.devices import DeviceStore, DeviceUnavailable
 from fluksio.flow.events import event_bus
 from fluksio.flow.executor import ExecutionService
 from fluksio.flow.metrics import MetricsCollector
@@ -538,6 +538,12 @@ async def invalid_name(request: Request, exc: InvalidName) -> JSONResponse:  # n
     return JSONResponse(status_code=400, content={"detail": str(exc)})
 
 
+@app.exception_handler(DeviceUnavailable)
+async def device_unavailable(request: Request, exc: DeviceUnavailable) -> JSONResponse:  # noqa: ARG001
+    """A device credential that cannot be checked now is not a refused one."""
+    return JSONResponse(status_code=503, content={"detail": str(exc)})
+
+
 @app.exception_handler(RebuildBusy)
 async def rebuild_busy(request: Request, exc: RebuildBusy) -> JSONResponse:  # noqa: ARG001
     """Every route that deploys something answers a wedged rebuild the same way.

backend/tests/api/routes/test_devices.py

@@ -2,7 +2,9 @@
 
 import threading
 
+import pytest
 from fastapi.testclient import TestClient
+from starlette.websockets import WebSocketDisconnect
 
 from fluksio.api.routes import devices as devices_route
 from fluksio.cloud import config as cloud_config
@@ -1053,6 +1055,64 @@ def test_unpairing_a_screen_leaves_its_panel_standing(
     assert client.post(f"{PREFIX}/kitchen/unpair").status_code == 401
 
 
+def test_a_device_is_refused_only_when_its_credential_is_gone(
+    client: TestClient,
+    superuser_token_headers: dict[str, str],
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """ "Cannot say right now" is no refusal: a refused device pairs again."""
+    _panels(client, superuser_token_headers, {"panels": [{"id": "hallway"}]})
+    device = _pair(client, superuser_token_headers, "hallway")
+    socket = f"{settings.API_V1_STR}/flows/ws"
+    store = app.state.device_store
+
+    def unreadable(_: str) -> int:
+        raise ValueError("torn")
+
+    # The nonces do not read: the socket is accepted and told to try again
+    # later, and the device's own read answers 503.
+    monkeypatch.setattr(store, "nonce", unreadable)
+    with (
+        client.websocket_connect(socket, headers=device) as ws,
+        pytest.raises(WebSocketDisconnect) as later,
+    ):
+        ws.receive_json()
+    assert later.value.code == 1013
+    assert client.get(f"{PREFIX}/hallway", headers=device).status_code == 503
+    monkeypatch.undo()
+    with client.websocket_connect(socket, headers=device) as ws:
+        assert "values" in ws.receive_json()
+
+    # Unpaired, it is refused before it is accepted, a 403 on the wire.
+    client.post(f"{PREFIX}/hallway/unpair", headers=superuser_token_headers)
+    with (
+        pytest.raises(WebSocketDisconnect) as refused,
+        client.websocket_connect(socket, headers=device),
+    ):
+        pass
+    assert refused.value.code == 1008
+    assert client.get(f"{PREFIX}/hallway", headers=device).status_code == 401
+
+
+def test_only_a_device_paired_here_turns_touch_on_with_its_report(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    """One paired before reports for the first time with an update, its
+    switch set by hand: a knob left off on purpose stays off."""
+    _panels(client, superuser_token_headers, {"panels": [{"id": "knob_old"}]})
+    device = _pair(client, superuser_token_headers, "knob_old")
+    store = app.state.device_store
+    with store._state_lock:
+        states = store._read_states()
+        states["knob_old"].paired_at = None
+        store._write_states(states)
+
+    put = client.put(f"{PREFIX}/knob_old/report", headers=device, json=REPORT)
+    assert put.status_code == 200, put.text
+    read = client.get(f"{PREFIX}/knob_old", headers=device).json()
+    assert read["touch"] is False
+
+
 class _FakeRedis:
     """Just enough Redis for the pending-code store: strings and one zset.
 

backend/tests/flow/test_device_node.py

@@ -3,8 +3,10 @@
 import asyncio
 from pathlib import Path
 
-from fluksio.flow.controller import FlowController
+from fluksio.flow.alerts import Alert, AlertManager, AlertsConfig, Channel, Rule
+from fluksio.flow.controller import HEALTH_FLOOR, FlowController
 from fluksio.flow.devices import presence
+from fluksio.flow.events import EventBus
 from fluksio.flow.messages import DType, MessageSpec
 from fluksio.flow.schemas import FlowDef, FlowInput, NodeDef
 from fluksio.flow.store import FlowStore
@@ -115,3 +117,51 @@ def test_an_offline_device_is_a_warning_not_a_fault(tmp_path: Path) -> None:
         ["hall.panel"],
         [("node_degraded", True)],
     ]
+
+
+def test_a_device_away_after_a_restart_alerts_nobody(tmp_path: Path) -> None:
+    """No push, and degraded on screen only once the floor has passed.
+
+    A restart starts every device node without its device, which comes back
+    within seconds, and a rule on every node's health is what sends phones a
+    push.
+    """
+    controller = _controller(tmp_path)
+    events: list[dict] = []
+    original = controller._publish
+    controller._publish = lambda event: (events.append(event), original(event))  # type: ignore[method-assign]
+    sent: list[Alert] = []
+    alerts = AlertManager(
+        EventBus(),
+        AlertsConfig(
+            channels=[Channel(name="phone", kind="ntfy", config={"topic": "t"})],
+            rules=[Rule(events=["node_health"], channels=["phone"])],
+        ),
+    )
+
+    async def capture(channel: Channel, alert: Alert) -> None:
+        sent.append(alert)
+
+    alerts.send = capture  # type: ignore[method-assign]
+
+    async def scenario() -> list[list[str]]:
+        await controller.reload()  # the restart: the device is not back yet
+        surfaced = [[e.id for e in controller.unhealthy_nodes(health="degraded")]]
+        presence.opened("wall")
+        presence.closed("wall")  # and it drops again
+        surfaced.append([e.id for e in controller.unhealthy_nodes(health="degraded")])
+        controller.loaded["hall.panel"].health_since -= HEALTH_FLOOR
+        surfaced.append([e.id for e in controller.unhealthy_nodes(health="degraded")])
+        for event in events:
+            await alerts.handle(event)
+        await controller._teardown()
+        return surfaced
+
+    assert asyncio.run(scenario()) == [[], [], ["hall.panel"]]
+    assert [e["health"] for e in events if e["type"] == "node_health"] == [
+        "degraded",
+        "ok",
+        "degraded",
+    ]
+    assert sent == []
+    presence.forget("wall")

backend/tests/flow/test_store.py

@@ -193,3 +193,14 @@ def test_opening_the_store_renames_a_nodes_device_to_worker(store: FlowStore):
         "prefer",
     )
     assert published.version == 2
+
+
+def test_a_client_from_before_the_rename_still_labels_its_workers():
+    """An older `fluksio sync` or SDK sends `device`, read as `worker` for a release."""
+    node = NodeDef.model_validate(
+        {"id": "fit", "device": "gpu", "device_policy": "prefer"}
+    )
+    assert (node.worker, node.worker_policy) == ("gpu", "prefer")
+    # Beside the new key, the new key wins.
+    both = NodeDef.model_validate({"id": "fit", "device": "cpu", "worker": "gpu"})
+    assert both.worker == "gpu"

docs/interface/dashboards.md

@@ -565,7 +565,8 @@ sends only while somebody wants it.
 The device's page shows the report under **Capabilities**, with each port's
 binding or **Use in flow** for it. A reported screen becomes a canvas preset in
 the dashboard settings, named after the device, and the first report of a
-touch screen turns **Touch friendly** on.
+touch screen turns **Touch friendly** on. A device paired before this version
+keeps the switch as it was set.
 
 ### Before device nodes