From 8df39030c21c17bd0aca879c940d76e0687fe216 Mon Sep 17 00:00:00 2001 From: xmap <16776958+xmap@users.noreply.github.com> Date: Mon, 10 Aug 2026 11:20:48 -0500 Subject: [PATCH] Record whether CORA reached the enclosure permit substrate The permit-status projection advances only on a change, so a stale value has meant "no transition since" and "not observed since" the same way: a silent gap in coverage reads as nothing happened. This adds a permit probe trail, entries_enclosure_permit_probes, one append-only row per observation the monitor's EnclosureObserver surfaces, separate from the EnclosurePermitObserved record of what the interlock said. reach_tier (RELAYED/UNREACHED) records whether CORA reached the configured channel this tick; a status-bearing push or disconnect still drives the permit transition as before, while a new periodic poll writes probe-only rows (no status claim, never a transition) so a quiet, unchanging PV doesn't read as a coverage gap. A stronger tier for a confirmed direct round trip is deliberately not shipped: no producer here can prove one yet (2-BM reads through a caching gateway), and an unearned strong claim is worse than none. The probe write can never suppress the real permit transition, and no row is written at all while a process boots in degraded schema mode, since the event store is read-only there and a dense trail would misrepresent a window CORA could not actually record. Went through three review rounds (design, re-cut, and a pass against the built code) that caught a startup-race regression, a route that could have let a stale poll read flip a live permit status, and a test double gap that silently turned a cancellation test into an infinite loop. Also adds a fitness test closing the same missing-GRANT class of bug already latent on five older entries_ tables (tracked separately; not fixed here). Item 3 of the 2-BM coverage-window commissioning ladder. The polling cadence defaults off (enclosure_permit_probe_tick_seconds=None) pending staff confirmation, since it would read the PSS gateway on a timer. Co-Authored-By: Claude Sonnet 5 --- .../cora/api/_enclosure_permit_observer.py | 120 ++++++- apps/api/src/cora/api/main.py | 2 + apps/api/src/cora/enclosure/_monitor.py | 130 +++++++- .../aggregates/enclosure/__init__.py | 12 + .../aggregates/enclosure/permit_probes.py | 130 ++++++++ .../enclosure/ports/enclosure_observer.py | 57 +++- apps/api/src/cora/enclosure/wire.py | 16 + apps/api/src/cora/infrastructure/config.py | 19 +- .../src/cora/infrastructure/schema_version.py | 2 +- .../architecture/test_entries_table_grants.py | 98 ++++++ .../test_enclosure_permit_monitor.py | 310 ++++++++++++++++-- .../api/test_enclosure_permit_observer.py | 159 ++++++++- .../unit/enclosure/test_enclosure_observer.py | 16 +- docs/architecture/modules/enclosure/index.md | 39 +++ ...0_init_entries_enclosure_permit_probes.sql | 83 +++++ infra/atlas/migrations/atlas.sum | 3 +- 16 files changed, 1124 insertions(+), 72 deletions(-) create mode 100644 apps/api/src/cora/enclosure/aggregates/enclosure/permit_probes.py create mode 100644 apps/api/tests/architecture/test_entries_table_grants.py create mode 100644 infra/atlas/migrations/20260810000000_init_entries_enclosure_permit_probes.sql diff --git a/apps/api/src/cora/api/_enclosure_permit_observer.py b/apps/api/src/cora/api/_enclosure_permit_observer.py index b3be51f641c..ba798f306ca 100644 --- a/apps/api/src/cora/api/_enclosure_permit_observer.py +++ b/apps/api/src/cora/api/_enclosure_permit_observer.py @@ -17,6 +17,30 @@ gate closed rather than leaving a stale `Permitted`. That synthesized observation carries NO substrate time, because there was no substrate reading behind it; see `_unknown`. + +## Permit probe trail: a sibling poller, not an in-pump interleave + +When `tick_seconds` is configured, each PV also gets a sibling polling +task (`_poll`) alongside its push subscription (`_pump`), both feeding +the same queue. The poll never carries a status claim +(`observed_status=None`): it exists only to re-affirm reach on a fixed +cadence, independent of push traffic, because EPICS CA monitors are +change-only and a quiet permit PV would otherwise leave a probe-trail +gap that is really coverage, not an outage. See +[[project_enclosure_permit_probe_design]]. + +The poller is a SIBLING of `_pump`, not nested inside it, deliberately: +`_pump` returns as soon as its subscription ends (clean end or +disconnect), and `_drain` only re-subscribes after every pump has +returned. A poller living inside `_pump` would die with it and could +never observe a PV's recovery. As a sibling it keeps polling through a +dead push path and its `UNREACHED` probes are the only signal left that +this specific PV, not the whole deployment, is unreachable. + +Because the poll never carries a status, it cannot drive a permit +transition and cannot be confused for evidence stronger than it is: a +successful poll proves only that the configured channel answered this +tick, never that the underlying signal is current (see `ReachTier`). """ from __future__ import annotations @@ -27,6 +51,7 @@ from cora.enclosure.ports.enclosure_observer import ( EnclosureObservation, EnclosureObserverScope, + ReachTier, ) from cora.operation.ports.control_port import ControlNotConnectedError, Measurement @@ -129,9 +154,11 @@ def __init__( *, control_port: ControlPort, permit_pvs: Mapping[str, str], + tick_seconds: float | None = None, ) -> None: self._control_port = control_port self._permit_pvs = dict(permit_pvs) + self._tick_seconds = tick_seconds def observe(self, scope: EnclosureObserverScope) -> AsyncGenerator[EnclosureObservation]: return self._drain(scope) @@ -145,8 +172,16 @@ async def _drain(self, scope: EnclosureObserverScope) -> AsyncGenerator[Enclosur if not pvs: return queue: asyncio.Queue[EnclosureObservation | _PumpDone] = asyncio.Queue() - tasks = [asyncio.create_task(self._pump(code, pv, queue)) for code, pv in pvs] - remaining = len(tasks) + pump_tasks = [asyncio.create_task(self._pump(code, pv, queue)) for code, pv in pvs] + poll_tasks = ( + [asyncio.create_task(self._poll(code, pv, queue)) for code, pv in pvs] + if self._tick_seconds is not None + else [] + ) + tasks = pump_tasks + poll_tasks + # Only pumps ever signal completion; a poller runs until the + # `finally` below cancels it, so it must not hold this open. + remaining = len(pump_tasks) try: while remaining > 0: item = await queue.get() @@ -154,6 +189,20 @@ async def _drain(self, scope: EnclosureObserverScope) -> AsyncGenerator[Enclosur remaining -= 1 continue yield item + # Every pump has finished, but a still-running poller can have + # enqueued a probe in the same instant the final _PumpDone was + # read (asyncio.Queue.put_nowait needs no await, so it is not + # ordered against the `remaining` check above). Drain exactly + # what is ALREADY queued right now, synchronously, into a list + # before yielding any of it: yielding suspends this generator + # and hands control back to a poller, which could otherwise + # keep queue.empty() perpetually False and stop `_drain` from + # ever returning to let the outer loop reconnect. + pending = queue.qsize() + leftover = [queue.get_nowait() for _ in range(pending)] + for item in leftover: + if not isinstance(item, _PumpDone): + yield item finally: for task in tasks: task.cancel() @@ -167,9 +216,19 @@ async def _pump( ) -> None: try: async for reading in self._control_port.subscribe(pv): + # RELAYED unconditionally: a delivered reading is push + # contact regardless of its mapped status. A Bad-quality + # reading still maps to "Unknown" but is NOT the same + # fact as `_unknown`'s disconnect: the substrate spoke + # and said its value could not be believed, which is + # reach with an unbelievable value, not absence of reach. queue.put_nowait( self._observation( - code, pv, permit_status_from_reading(reading), reading.produced_at + code, + pv, + permit_status_from_reading(reading), + reading.produced_at, + reach_tier=ReachTier.RELAYED, ) ) # Clean stream end: permit becomes Unknown until re-subscribed. @@ -179,17 +238,64 @@ async def _pump( finally: queue.put_nowait(_PUMP_DONE) + async def _poll( + self, + code: str, + pv: str, + queue: asyncio.Queue[EnclosureObservation | _PumpDone], + ) -> None: + """Re-affirm reach to `pv` every `_tick_seconds`, independent of push. + + Never pushes `_PumpDone`: this task is a sibling of `_pump`, not + a stage in its lifecycle, and runs until `_drain`'s `finally` + cancels it on teardown. It ticks unconditionally, regardless of + how much push traffic `pv` is producing, which is simpler than + gating on push quiescence and avoids the "a chatty PV is never + polled" surprise a quiescence-gated poll would carry. + + A tick that fails (any exception but cancellation) writes an + `UNREACHED` probe-only observation and keeps polling; it never + raises out of this loop and never touches `_pump`'s subscription. + """ + assert self._tick_seconds is not None + while True: + await asyncio.sleep(self._tick_seconds) + try: + await self._control_port.read(pv) + except Exception: # any read failure is a failed probe, not a bug + queue.put_nowait(self._probe_only(code, pv, ReachTier.UNREACHED)) + else: + queue.put_nowait(self._probe_only(code, pv, ReachTier.RELAYED)) + def _observation( - self, code: str, pv: str, status: str, observed_at: datetime | None + self, + code: str, + pv: str, + status: str, + observed_at: datetime | None, + *, + reach_tier: ReachTier, ) -> EnclosureObservation: return EnclosureObservation( enclosure_code=code, observed_status=status, + reach_tier=reach_tier, observed_at=observed_at, source_kind=_SOURCE_KIND, source_id=pv, ) + def _probe_only(self, code: str, pv: str, reach_tier: ReachTier) -> EnclosureObservation: + """A poll tick's result: reach evidence with no status claim.""" + return EnclosureObservation( + enclosure_code=code, + observed_status=None, + reach_tier=reach_tier, + observed_at=None, + source_kind=_SOURCE_KIND, + source_id=pv, + ) + def _unknown(self, code: str, pv: str) -> EnclosureObservation: """A disconnect or stream end, which carries NO substrate time. @@ -209,9 +315,11 @@ def _unknown(self, code: str, pv: str) -> EnclosureObservation: nothing. The recording side keeps its own clock: the event's `occurred_at` - still says when CORA learned of the disconnect. + still says when CORA learned of the disconnect. `reach_tier` is + `UNREACHED`: a disconnect carries a status claim (`Unknown`, so + the run gate still fails closed) but is not reach evidence. """ - return self._observation(code, pv, _UNKNOWN, None) + return self._observation(code, pv, _UNKNOWN, None, reach_tier=ReachTier.UNREACHED) __all__ = ["ControlPortEnclosureObserver", "permit_status_from_reading"] diff --git a/apps/api/src/cora/api/main.py b/apps/api/src/cora/api/main.py index 6ae68ddf2c7..858b85b8fca 100644 --- a/apps/api/src/cora/api/main.py +++ b/apps/api/src/cora/api/main.py @@ -1009,6 +1009,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: enclosure_permit_observer = ControlPortEnclosureObserver( control_port=app.state.operation.control_port, permit_pvs=settings.enclosure_permit_pvs, + tick_seconds=settings.enclosure_permit_probe_tick_seconds, ) try: @@ -1019,6 +1020,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None]: observer=enclosure_permit_observer, kernel=deps, name_to_id=enclosure_permit_ids, + probe_store=app.state.enclosure.permit_probe_store, enclosure_projection_registry=enclosure_only_registry, startup_timeout_seconds=( settings.enclosure_permit_monitor_startup_timeout_seconds diff --git a/apps/api/src/cora/enclosure/_monitor.py b/apps/api/src/cora/enclosure/_monitor.py index 185ce41713c..d80153a935d 100644 --- a/apps/api/src/cora/enclosure/_monitor.py +++ b/apps/api/src/cora/enclosure/_monitor.py @@ -57,6 +57,33 @@ settlement happened, never that no more are pending, and a PV that never answers leaves that one enclosure's preflight reading exactly as stale as before this fix. + +## Permit probe trail + +Every observation the observer surfaces, whether or not it carries a +status claim, is also recorded as a `PermitProbe` row: an append-only +fact about whether CORA reached the permit substrate, kept separate from +`EnclosurePermitObserved`'s record of what the interlock said. See +[[project_enclosure_permit_probe_design]]. The probe write happens in +its own try/except, before the transition attempt, so a bookkeeping +failure there can never suppress a real permit transition (the reverse +of "Startup race" above: this failure mode makes the record LESS +truthful, not the boot LESS available, so it degrades by logging and +continuing rather than by warning and proceeding). + +`observation.observed_status is None` means the observation is +probe-only (a periodic re-affirmation read that intentionally makes no +status claim): the probe row is still written, but no permit transition +is attempted and no startup-readiness code is settled by it (see +`run_enclosure_permit_monitor`), because it proves nothing about +whether the enclosure's actual permit-status has been confirmed. + +No probe row is written at all while `kernel.schema_posture == +"degraded"` (a boot running under `ALLOW_SCHEMA_VERSION_MISMATCH`, +whose event store is read-only): a probe row asserting reach during a +window where CORA cannot actually record what it observed would be +worse than silence. The resulting gap in the trail is the correct +signal, not a bug. """ from __future__ import annotations @@ -71,6 +98,7 @@ EnclosureEvent, EnclosurePermitStatus, MonitorRef, + PermitProbe, event_type_name, fold, from_stored, @@ -88,6 +116,7 @@ if TYPE_CHECKING: from collections.abc import AsyncGenerator, Mapping + from cora.enclosure.aggregates.enclosure import PermitProbeStore from cora.enclosure.ports.enclosure_observer import ( EnclosureObservation, EnclosureObserver, @@ -117,16 +146,68 @@ async def record_observation( kernel: Kernel, observation: EnclosureObservation, name_to_id: Mapping[str, UUID], + probe_store: PermitProbeStore, ) -> None: - """Record one observation as an EnclosurePermitObserved (raw, authz-bypassed). - - No-op when the code is unmapped, the status is unparseable, or the - decider returns `[]` (identical-status, status-change-only). + """Record one observation: a permit-probe row, then, when status-bearing, + an EnclosurePermitObserved transition (raw, authz-bypassed). + + No-op entirely when the code is unmapped: the row cannot be + attributed to an enclosure. Otherwise the probe row is written + unconditionally (except see the degraded-schema case below), in its + own try/except: a probe-store failure must never suppress the + transition below it (a bookkeeping table must never take down the + safety-relevant record it exists to annotate). + + `kernel.schema_posture == "degraded"` skips the probe write entirely + rather than writing one: a degraded boot runs a read-only event + store (see `Kernel.schema_posture`), so a probe row asserting reach + during that window would claim coverage over a process that cannot + actually record what it observed. A GAP in the trail here is the + correct signal, not a bug: it is exactly the "CORA was not really + watching" fact the trail exists to preserve, and writing a row would + hide it behind a dense, misleading RELAYED/UNREACHED history. + + `observation.observed_status is None` means this observation is + probe-only and makes no status claim, so no transition is attempted + past the probe write. Otherwise, an unparseable status, or the + decider returning `[]` (identical-status, status-change-only), are + no-ops on the transition path only; the probe row still stands. """ enclosure_id = name_to_id.get(observation.enclosure_code) if enclosure_id is None: _log.warning("enclosure_monitor.unknown_code", enclosure_code=observation.enclosure_code) return + + if kernel.schema_posture == "degraded": + _log.warning( + "enclosure_monitor.probe_skipped_degraded_schema", + enclosure_code=observation.enclosure_code, + ) + else: + try: + await probe_store.append( + [ + PermitProbe( + event_id=kernel.id_generator.new_id(), + enclosure_id=enclosure_id, + source_kind=observation.source_kind, + source_id=observation.source_id, + reach_tier=observation.reach_tier, + status_claimed=observation.observed_status is not None, + ) + ] + ) + except asyncio.CancelledError: + raise + except Exception: + _log.exception( + "enclosure_monitor.probe_write_failed", + enclosure_code=observation.enclosure_code, + ) + + if observation.observed_status is None: + return + try: new_status = EnclosurePermitStatus(observation.observed_status) except ValueError: @@ -188,18 +269,32 @@ async def run_enclosure_permit_monitor( observer: EnclosureObserver, kernel: Kernel, name_to_id: Mapping[str, UUID], + probe_store: PermitProbeStore, reconnect_delay_seconds: float = _RECONNECT_DELAY_SECONDS, startup_ready: asyncio.Event | None = None, ) -> None: """Drain the observer, recording each observation; re-subscribe on stream end. - `startup_ready`, when given, is set as soon as every configured enclosure - code has produced one observation (a real reading or the observer's own - `Unknown` on a dead PV), or, failing that, once the current pass ends (the - observer's stream terminated or raised with codes still unheard from). The - second case fires on every reconnect attempt, not only the first, but - `Event.set` on an already-set event is a no-op, so the caller only ever - sees the earliest settlement. + `startup_ready`, when given, is set once every configured enclosure code + has produced one STATUS-BEARING observation (a real reading, or the + observer's own `Unknown` on a dead PV; `observation.observed_status is + not None`). A probe-only observation (a periodic re-affirmation poll, + `observed_status is None`) never settles a code: it proves reach, not + that the enclosure's permit status has been confirmed, and settling on + it would let a poll-only pass mark boot ready while the actual permit + transition attempt (which only a status-bearing observation can make) + never happened, serving a stale `permit_status` boot never confirmed. + + Failing that, `startup_ready` is set once a full pass ends (the + observer's stream terminated or raised) with every code still settled + from a PRIOR pass, i.e. `pending_codes` is already empty; a pass that + ends with codes still pending (a total connection failure, or the + observer producing nothing at all) does NOT settle `startup_ready` + here, so the caller's own bounded wait (see `enclosure_permit_monitor_lifespan`) + is what gives up on a deployment that cannot connect at all, rather + than this loop falsely reporting readiness for a code it never heard + a status-bearing observation for. `Event.set` on an already-set event + is a no-op, so the caller only ever sees the earliest settlement. """ if not name_to_id: return @@ -209,7 +304,7 @@ async def run_enclosure_permit_monitor( try: async for observation in observer.observe(scope): try: - await record_observation(kernel, observation, name_to_id) + await record_observation(kernel, observation, name_to_id, probe_store) except asyncio.CancelledError: raise except Exception: @@ -217,14 +312,15 @@ async def run_enclosure_permit_monitor( "enclosure_monitor.record_failed", enclosure_code=observation.enclosure_code, ) - pending_codes.discard(observation.enclosure_code) - if startup_ready is not None and not pending_codes: - startup_ready.set() + if observation.observed_status is not None: + pending_codes.discard(observation.enclosure_code) + if startup_ready is not None and not pending_codes: + startup_ready.set() except asyncio.CancelledError: raise except Exception: _log.exception("enclosure_monitor.iteration_failed") - if startup_ready is not None: + if startup_ready is not None and not pending_codes: startup_ready.set() await asyncio.sleep(reconnect_delay_seconds) @@ -235,6 +331,7 @@ async def enclosure_permit_monitor_lifespan( observer: EnclosureObserver, kernel: Kernel, name_to_id: Mapping[str, UUID], + probe_store: PermitProbeStore, enclosure_projection_registry: ProjectionRegistry | None = None, startup_timeout_seconds: float = _STARTUP_TIMEOUT_SECONDS, ) -> AsyncGenerator[None]: @@ -266,6 +363,7 @@ async def enclosure_permit_monitor_lifespan( observer=observer, kernel=kernel, name_to_id=name_to_id, + probe_store=probe_store, startup_ready=startup_ready, ), name="enclosure-permit-monitor", diff --git a/apps/api/src/cora/enclosure/aggregates/enclosure/__init__.py b/apps/api/src/cora/enclosure/aggregates/enclosure/__init__.py index 0915a0179a4..f076fce06fc 100644 --- a/apps/api/src/cora/enclosure/aggregates/enclosure/__init__.py +++ b/apps/api/src/cora/enclosure/aggregates/enclosure/__init__.py @@ -30,6 +30,13 @@ to_payload, ) from cora.enclosure.aggregates.enclosure.evolver import evolve, fold +from cora.enclosure.aggregates.enclosure.permit_probes import ( + InMemoryPermitProbeStore, + PermitProbe, + PermitProbeStore, + PostgresPermitProbeStore, + ReachTier, +) from cora.enclosure.aggregates.enclosure.state import ( ENCLOSURE_NAME_MAX_LENGTH, Enclosure, @@ -62,11 +69,16 @@ "EnclosurePermitStatus", "EnclosureReason", "EnclosureRegistered", + "InMemoryPermitProbeStore", "InvalidEnclosureNameError", "InvalidEnclosureReasonError", "InvalidMonitorRefError", "MonitorRef", "MonitorTriggerNotPermittedError", + "PermitProbe", + "PermitProbeStore", + "PostgresPermitProbeStore", + "ReachTier", "event_type_name", "evolve", "fold", diff --git a/apps/api/src/cora/enclosure/aggregates/enclosure/permit_probes.py b/apps/api/src/cora/enclosure/aggregates/enclosure/permit_probes.py new file mode 100644 index 00000000000..d11be3c5cc2 --- /dev/null +++ b/apps/api/src/cora/enclosure/aggregates/enclosure/permit_probes.py @@ -0,0 +1,130 @@ +"""Permit probe entry: append-only record of reach to the permit substrate. + +The write half of the coverage-window seam +([[project_enclosure_permit_probe_design]]). The Enclosure BC's permit +monitor writes one `PermitProbe` per observation its `EnclosureObserver` +surfaces, so the read side can tell "the hutch has genuinely been +secure for six hours" from "CORA was not looking for six hours and got +lucky". `permit_status` (on `proj_enclosure_summary`) answers what the +interlock said; this table answers whether CORA could reach it. Neither +substitutes for the other, and this module carries no permit value. + +Mirrors the `FeedHeartbeat` per-category-writer pattern: a typed +dataclass + a category-local Protocol + Postgres / InMemory adapters, +BC-internal (NOT a shared cross-BC port). Append-only INSERT: the +entries_* table is REVOKEd from UPDATE, and there is no natural key to +deduplicate against (`event_id` is a fresh id per observation), so +unlike `FeedHeartbeatStore` this store does not need `ON CONFLICT`. +""" + +# pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false + +from dataclasses import dataclass +from enum import StrEnum +from typing import Protocol +from uuid import UUID + +import asyncpg + + +class ReachTier(StrEnum): + """How CORA reached the permit substrate for one observation. + + Two values ship in v1. `RELAYED` means CORA received or fetched a + value through the configured channel; `UNREACHED` means it could + not, this tick. A stronger tier for a confirmed direct round trip to + the authoritative source (as opposed to an intermediary, such as an + EPICS CA gateway that may answer from its own cache) is deliberately + NOT defined here: no producer in this codebase can currently prove + one, and an unearned strong claim is worse than none. Adding a value + later needs no migration, since the column is a length-CHECK, not a + value-enumerating CHECK. + """ + + RELAYED = "Relayed" + UNREACHED = "Unreached" + + +@dataclass(frozen=True) +class PermitProbe: + """One reach observation for an Enclosure's permit substrate. + + `event_id` is the producer-assigned UUIDv7 dedup key (PK). `source_kind` + / `source_id` mirror the attribution pair on `MonitorRef` (the same + substrate, e.g. an EPICS PV). `status_claimed` records whether the + observation this probe accompanies also carried a permit-status claim + (a push delivery, or a real substrate disconnect) as opposed to being + probe-only (a periodic re-affirmation read that makes no status + claim). This is a fact about the PROBE, not the hutch: the probe row + never carries the observed permit value itself. `recorded_at` (DB + DEFAULT now()) is the trust anchor and is not carried on this row, no + producer-asserted timestamp exists to pair it with. + """ + + event_id: UUID + enclosure_id: UUID + source_kind: str + source_id: str + reach_tier: ReachTier + status_claimed: bool + + +class PermitProbeStore(Protocol): + """Per-category port for permit-probe writes (BC-internal).""" + + async def append(self, rows: list[PermitProbe]) -> None: ... + + +_APPEND_SQL = """ +INSERT INTO entries_enclosure_permit_probes ( + event_id, enclosure_id, source_kind, source_id, reach_tier, status_claimed +) VALUES ($1, $2, $3, $4, $5, $6) +""" + + +class PostgresPermitProbeStore: + """asyncpg-backed `PermitProbeStore`.""" + + def __init__(self, pool: asyncpg.Pool) -> None: + self._pool = pool + + async def append(self, rows: list[PermitProbe]) -> None: + if not rows: + return + async with self._pool.acquire() as conn: + await conn.executemany( + _APPEND_SQL, + [ + ( + r.event_id, + r.enclosure_id, + r.source_kind, + r.source_id, + r.reach_tier.value, + r.status_claimed, + ) + for r in rows + ], + ) + + +class InMemoryPermitProbeStore: + """Test / `app_env=test` adapter; list of every row appended.""" + + def __init__(self) -> None: + self._rows: list[PermitProbe] = [] + + async def append(self, rows: list[PermitProbe]) -> None: + self._rows.extend(rows) + + def all(self) -> list[PermitProbe]: + return list(self._rows) + + +__all__ = [ + "InMemoryPermitProbeStore", + "PermitProbe", + "PermitProbeStore", + "PostgresPermitProbeStore", + "ReachTier", +] diff --git a/apps/api/src/cora/enclosure/ports/enclosure_observer.py b/apps/api/src/cora/enclosure/ports/enclosure_observer.py index 09a758f0670..d8b0c826647 100644 --- a/apps/api/src/cora/enclosure/ports/enclosure_observer.py +++ b/apps/api/src/cora/enclosure/ports/enclosure_observer.py @@ -69,6 +69,9 @@ scope. Substrate severity codes are flattened by the adapter into the three-value `EnclosurePermitStatus` codomain before crossing this seam; severity bookkeeping belongs to the substrate, not the spine. +`reach_tier` (see `ReachTier`) is not a severity scalar in this sense: +it grades CORA's own reach to the substrate, never the hazard the +substrate reports. ## Stub roster @@ -95,7 +98,12 @@ from datetime import datetime from typing import Protocol, runtime_checkable -from cora.enclosure.aggregates.enclosure import EnclosurePermitStatus +# `ReachTier` is re-exported (see `__all__` below) rather than left for +# consumers to import from `cora.enclosure.aggregates.enclosure` +# directly: `cora.aggregates` submodules are tach-walled from the +# composition root (`cora.api`), while this port module is the blessed +# public surface, exactly like `EnclosureObservation` itself. +from cora.enclosure.aggregates.enclosure import EnclosurePermitStatus, ReachTier # No `_STUB_OBSERVED_AT`. The stub has no substrate behind it, so it # reports no substrate time. A fixed 1970 sentinel here would be a date @@ -106,7 +114,7 @@ @dataclass(frozen=True) class EnclosureObservation: - """One permit-status reading drained from the substrate. + """One reach observation from the permit substrate. `enclosure_code` is the BC-local Enclosure identity surface adapters know (the operator-readable code an EPICS PV's record @@ -115,10 +123,22 @@ class EnclosureObservation: decider. `observed_status` is the raw status string the adapter parsed - from the substrate. The decider parses it against the - `EnclosurePermitStatus` codomain and raises if the string does - not match a known status value; substrate values that cannot be - classified should be flattened to `"Unknown"` by the adapter. + from the substrate, or `None` when this observation makes no + status claim at all: a probe-only re-affirmation read that + exists to record reach, not to report a permit value. When not + `None`, the decider parses it against the `EnclosurePermitStatus` + codomain and raises if the string does not match a known status + value; substrate values that cannot be classified should be + flattened to `"Unknown"` by the adapter. A `None` status never + causes a permit transition, by construction: there is nothing to + parse. + + `reach_tier` says what kind of evidence backed this observation + (see `ReachTier`). It is required, not optional, so every adapter + states its own evidence rather than letting the recorder infer it + from `observed_status`, which cannot distinguish "the substrate + said Unknown" from "nothing was heard from the substrate at all" + (both would otherwise look identical). `observed_at` is the substrate's own time for the observation, and is `None` when the substrate supplied none. Real equipment @@ -134,21 +154,21 @@ class EnclosureObservation: indistinguishable from a reported one once it is written down. The recording side always has its own clock for the second fact. - NOTE: this field does not currently reach the event payload. - `_monitor.record_observation` builds `ObserveEnclosureStatus` - without it, so the recorded `EnclosurePermitObserved` carries - CORA's ingest time only. Threading it through the command, event - and projection is the next slice of - [[project-source-timestamp-design]]; until then, treat this as a - port-level fact the seam still drops. + `observed_at` reaches the event payload: `_monitor.record_observation` + threads it onto `ObserveEnclosureStatus`, and `EnclosurePermitObserved` + carries it alongside `occurred_at` (CORA's ingest time). This field + is untouched by the permit-probe trail, which records reach + separately and never carries a permit value. `source_kind` and `source_id` ship as separate strings; the handler joins them into the colon-delimited `monitor_ref` wire - string on the `EnclosurePermitObserved` event payload. + string on the `EnclosurePermitObserved` event payload, and the + same pair is carried unmodified onto the probe trail row. """ enclosure_code: str - observed_status: str + observed_status: str | None + reach_tier: ReachTier observed_at: datetime | None source_kind: str source_id: str @@ -212,6 +232,11 @@ class AlwaysPermittedEnclosureObserver: - `observed_status="Permitted"` matching `EnclosurePermitStatus.PERMITTED`. + - `reach_tier=ReachTier.RELAYED`, because the stub always + delivers a status claim, the same shape as a real push + delivery. `UNREACHED` would pair a delivered status with a + tier that means "no reach evidence", which is the illegal + combination the port docstring warns adapters against. - `observed_at=None`, because a stub has no substrate and so has no substrate time to report. This was a fixed 1970 sentinel, chosen for determinism, until the nullable field made honesty @@ -235,6 +260,7 @@ async def _drain(self, scope: EnclosureObserverScope) -> AsyncGenerator[Enclosur yield EnclosureObservation( enclosure_code=enclosure_code, observed_status=EnclosurePermitStatus.PERMITTED.value, + reach_tier=ReachTier.RELAYED, observed_at=None, source_kind="Stub", source_id="AlwaysPermittedEnclosureObserver", @@ -246,4 +272,5 @@ async def _drain(self, scope: EnclosureObserverScope) -> AsyncGenerator[Enclosur "EnclosureObservation", "EnclosureObserver", "EnclosureObserverScope", + "ReachTier", ] diff --git a/apps/api/src/cora/enclosure/wire.py b/apps/api/src/cora/enclosure/wire.py index 4f20e690e3b..9ebae54de6d 100644 --- a/apps/api/src/cora/enclosure/wire.py +++ b/apps/api/src/cora/enclosure/wire.py @@ -34,6 +34,11 @@ from dataclasses import dataclass from uuid import UUID +from cora.enclosure.aggregates.enclosure import ( + InMemoryPermitProbeStore, + PermitProbeStore, + PostgresPermitProbeStore, +) from cora.enclosure.features import ( decommission_enclosure, observe_enclosure_status, @@ -53,11 +58,22 @@ class EnclosureHandlers: register_enclosure: register_enclosure.IdempotentHandler observe_enclosure_status: observe_enclosure_status.Handler decommission_enclosure: decommission_enclosure.Handler + permit_probe_store: PermitProbeStore + """The permit-probe trail's write store. Surfaced on the bundle, + not a handler, so the FastAPI lifespan can hand it to the permit + monitor at `enclosure_permit_monitor_lifespan`'s call site (mirrors + `OperationHandlers.control_port`, surfaced the same way for the + same reason: a composition-root lifespan needs a dependency that + isn't itself a command handler).""" def wire_enclosure(deps: Kernel) -> EnclosureHandlers: """Build the Enclosure BC handlers from shared dependencies.""" + permit_probe_store: PermitProbeStore = ( + PostgresPermitProbeStore(deps.pool) if deps.pool is not None else InMemoryPermitProbeStore() + ) return EnclosureHandlers( + permit_probe_store=permit_probe_store, register_enclosure=with_tracing( with_idempotency( register_enclosure.bind(deps), diff --git a/apps/api/src/cora/infrastructure/config.py b/apps/api/src/cora/infrastructure/config.py index f652b72c8cb..0b47d8ee365 100644 --- a/apps/api/src/cora/infrastructure/config.py +++ b/apps/api/src/cora/infrastructure/config.py @@ -623,9 +623,26 @@ class Settings(BaseSettings): # # The keys are the enclosures to seed (under self_facility_code) and # monitor; the values are their SecureM PVs. See - # `cora.enclosure.adapters.control_port_enclosure_observer`. + # `cora.api._enclosure_permit_observer`. enclosure_permit_pvs: dict[str, str] = {} + # Permit probe trail (coverage-window commissioning ladder item 3; + # [[project_enclosure_permit_probe_design]]). Bounds how often + # `ControlPortEnclosureObserver` re-reads each configured permit PV + # independent of push traffic, so a quiet PV (EPICS CA monitors are + # change-only) does not leave a probe-trail gap that reads as a + # coverage outage when CORA was in fact still watching. `None` + # (default) disables polling entirely: the trail is then push-only, + # written only when the substrate itself sends something. This is an + # OPERATIONAL KILL SWITCH, not just a test-determinism convenience: + # setting it back to `None` stops the periodic read against the PSS + # gateway (a shared facility resource) without any other code change, + # and is the correct rollback for this feature if the poll cadence + # ever needs revisiting. Confirm a cadence with beamline staff before + # setting this in a deployment: it polls a facility resource on a + # timer. Irrelevant when `enclosure_permit_pvs` is empty. + enclosure_permit_probe_tick_seconds: float | None = None + # Bounds how long boot waits for the permit monitor's first settled read # per configured enclosure before serving requests (the startup-race # window documented in `cora.enclosure._monitor`, "Startup race"). A PV diff --git a/apps/api/src/cora/infrastructure/schema_version.py b/apps/api/src/cora/infrastructure/schema_version.py index 80fae160cb1..0829c609028 100644 --- a/apps/api/src/cora/infrastructure/schema_version.py +++ b/apps/api/src/cora/infrastructure/schema_version.py @@ -74,7 +74,7 @@ class SchemaCheck: expected: str -EXPECTED_SCHEMA_VERSION: Final = "20260809190000" +EXPECTED_SCHEMA_VERSION: Final = "20260810000000" """The newest migration this build was written against. Hand-maintained, and deliberately not derived at runtime: the image does diff --git a/apps/api/tests/architecture/test_entries_table_grants.py b/apps/api/tests/architecture/test_entries_table_grants.py new file mode 100644 index 00000000000..be96ae68b06 --- /dev/null +++ b/apps/api/tests/architecture/test_entries_table_grants.py @@ -0,0 +1,98 @@ +"""Every `entries_*` / `events` table created in migrations carries a +matching GRANT on cora_app. + +Sibling of `test_projection_grants.py` (proj_* tables) and +`test_migration_revokes.py` (the REVOKE half of the same append-only +tables this file checks). `entries_*` migration headers have +historically claimed cora_app "gets SELECT + INSERT via ALTER DEFAULT +PRIVILEGES" (see e.g. `20260621040000_init_entries_run_feed_heartbeats.sql`); +that claim is FALSE for tables (the role-init migration's +`ALTER DEFAULT PRIVILEGES` covers sequences only, per +`20260512230000_init_role_cora_app.sql`), so several tables created +this way carry no working grant at all. `_GRANDFATHERED` lists the +tables already affected; fixing them is a separate, already-tracked +follow-up (a GRANT-only migration against a live production database), +not something to silently paper over here. This test's job is to make +sure the mistake stops recurring: every table NOT on that list must +carry an explicit GRANT. +""" + +from __future__ import annotations + +import re + +import pytest + +from tests.architecture.conftest import tracked_migration_files + +_CREATE_TABLE_RE = re.compile( + r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?([a-zA-Z_][a-zA-Z0-9_]*)", + re.IGNORECASE, +) + +# Tables confirmed (2026-08-10, alongside the enclosure permit probe +# trail's own migration review) to rely on the false ALTER DEFAULT +# PRIVILEGES claim and carry no working GRANT today. Do not add to this +# list going forward: a new table belongs in a migration with its own +# explicit GRANT, per the assertion message below. +_GRANDFATHERED = frozenset( + { + "entries_run_readings", + "entries_operation_procedure_steps", + "entries_run_feed_heartbeats", + "entries_operation_procedure_diagnostics", + "entries_operation_procedure_outcomes", + } +) + + +def _all_migration_text() -> str: + return "\n".join(f.read_text() for f in tracked_migration_files()) + + +def _append_only_tables_created() -> set[str]: + out: set[str] = set() + for path in tracked_migration_files(): + for match in _CREATE_TABLE_RE.finditer(path.read_text()): + name = match.group(1) + if name == "events" or name.startswith("entries_"): + out.add(name) + return out + + +@pytest.mark.architecture +def test_every_new_entries_table_has_cora_app_grant() -> None: + """Pattern accepted: `GRANT ... ON [TABLE] ... TO cora_app`. + + Tables on `_GRANDFATHERED` are skipped: they predate this test and + fixing them is a separate production migration, not a test change. + A table that is renamed off the grandfathered list (its CREATE TABLE + name changes) is NOT exempt under its new name; only the exact + grandfathered identifiers are excused. + """ + haystack = _all_migration_text() + tables = _append_only_tables_created() - _GRANDFATHERED + assert tables, ( + "No non-grandfathered append-only tables found; either the schema " + "is empty, table-name detection is wrong, or _GRANDFATHERED has " + "swallowed every table (check it against tracked_migration_files())." + ) + + missing: list[str] = [] + for table in sorted(tables): + pattern = re.compile( + rf"GRANT\b[^;]*\bON\s+(?:TABLE\s+)?{re.escape(table)}\b[^;]*\bTO\s+[^;]*cora_app\b", + re.IGNORECASE | re.DOTALL, + ) + if not pattern.search(haystack): + missing.append(table) + + assert not missing, ( + "entries_* / events tables missing a GRANT on cora_app:\n" + + "\n".join(f" - {t}" for t in missing) + + "\n\nAdd a `GRANT SELECT, INSERT ON
TO cora_app;` statement " + "to the migration that creates the table. Do NOT add the table to " + "_GRANDFATHERED instead: that list is closed to the tables already " + "affected by the ALTER DEFAULT PRIVILEGES mistake this test exists " + "to stop repeating." + ) diff --git a/apps/api/tests/integration/test_enclosure_permit_monitor.py b/apps/api/tests/integration/test_enclosure_permit_monitor.py index 24ed5118320..f2af45b4b67 100644 --- a/apps/api/tests/integration/test_enclosure_permit_monitor.py +++ b/apps/api/tests/integration/test_enclosure_permit_monitor.py @@ -8,6 +8,13 @@ startup-race fix (module docstring "Startup race") is covered separately: `startup_ready` semantics on the loop, and the lifespan's bounded wait plus its still-yields-on-timeout fallback. + +The permit probe trail (module docstring "Permit probe trail", +[[project_enclosure_permit_probe_design]]) is covered separately below: +every mapped observation writes a probe row before any transition is +attempted; a probe-only observation (`observed_status=None`) writes a +row but never touches the event store and never settles startup +readiness; a probe-store failure never suppresses a real transition. """ # pyright: reportUnknownMemberType=false, reportUnknownVariableType=false, reportUnknownArgumentType=false @@ -31,6 +38,12 @@ run_enclosure_permit_monitor, ) from cora.enclosure.adapters import PostgresEnclosureLookup +from cora.enclosure.aggregates.enclosure import ( + InMemoryPermitProbeStore, + PermitProbe, + PostgresPermitProbeStore, + ReachTier, +) from cora.enclosure.ports.enclosure_observer import ( EnclosureObservation, EnclosureObserverScope, @@ -53,10 +66,17 @@ def _deps_with(db_pool: asyncpg.Pool, *, permit_pvs: dict[str, str]) -> Kernel: ) -def _obs(name: str, status: str, *, pv: str = "S02BM-PSS:StaA:SecureM") -> EnclosureObservation: +def _obs( + name: str, + status: str | None, + *, + pv: str = "S02BM-PSS:StaA:SecureM", + reach_tier: ReachTier = ReachTier.RELAYED, +) -> EnclosureObservation: return EnclosureObservation( enclosure_code=name, observed_status=status, + reach_tier=reach_tier, observed_at=_T, source_kind="EpicsPv", source_id=pv, @@ -74,7 +94,7 @@ async def test_record_observation_writes_permit_observed(db_pool: asyncpg.Pool) deps = _deps_with(db_pool, permit_pvs={name: "pv"}) name_to_id = await seed_enclosures(deps) - await record_observation(deps, _obs(name, "Permitted"), name_to_id) + await record_observation(deps, _obs(name, "Permitted"), name_to_id, InMemoryPermitProbeStore()) events = await _permit_events(deps, name_to_id[name]) assert len(events) == 1 @@ -83,39 +103,60 @@ async def test_record_observation_writes_permit_observed(db_pool: asyncpg.Pool) @pytest.mark.integration -async def test_record_observation_same_status_is_noop(db_pool: asyncpg.Pool) -> None: +async def test_record_observation_same_status_writes_probes_but_no_second_event( + db_pool: asyncpg.Pool, +) -> None: name = f"hutch-idem-{uuid4().hex[:8]}" deps = _deps_with(db_pool, permit_pvs={name: "pv"}) name_to_id = await seed_enclosures(deps) + probe_store = InMemoryPermitProbeStore() - await record_observation(deps, _obs(name, "Permitted"), name_to_id) - await record_observation(deps, _obs(name, "Permitted"), name_to_id) + await record_observation(deps, _obs(name, "Permitted"), name_to_id, probe_store) + await record_observation(deps, _obs(name, "Permitted"), name_to_id, probe_store) + # The decider's status-change-only short-circuit still absorbs the + # second observation on the transition path... assert len(await _permit_events(deps, name_to_id[name])) == 1 + # ...but the probe trail is not the event stream: a probe row is + # written per observation regardless of whether it caused a + # transition, which is the entire point of this slice. + assert len(probe_store.all()) == 2 @pytest.mark.integration -async def test_record_observation_unknown_code_is_noop(db_pool: asyncpg.Pool) -> None: +async def test_record_observation_unknown_code_writes_no_probe_or_event( + db_pool: asyncpg.Pool, +) -> None: name = f"hutch-known-{uuid4().hex[:8]}" deps = _deps_with(db_pool, permit_pvs={name: "pv"}) name_to_id = await seed_enclosures(deps) + probe_store = InMemoryPermitProbeStore() - # observation for a code that was never seeded -> skipped, no raise - await record_observation(deps, _obs("not-a-hutch", "Permitted"), name_to_id) + # observation for a code that was never seeded -> skipped, no raise, + # and no probe row: an unmapped code cannot be attributed to an + # enclosure at all. + await record_observation(deps, _obs("not-a-hutch", "Permitted"), name_to_id, probe_store) assert await _permit_events(deps, name_to_id[name]) == [] + assert probe_store.all() == [] @pytest.mark.unit async def test_run_monitor_empty_map_returns_immediately() -> None: await run_enclosure_permit_monitor( - observer=_FakeObserver([]), kernel=cast("Kernel", None), name_to_id={} + observer=_FakeObserver([]), + kernel=cast("Kernel", None), + name_to_id={}, + probe_store=InMemoryPermitProbeStore(), ) @pytest.mark.unit async def test_lifespan_empty_map_is_noop() -> None: async with enclosure_permit_monitor_lifespan( - observer=_FakeObserver([]), kernel=cast("Kernel", None), name_to_id={} + observer=_FakeObserver([]), + kernel=cast("Kernel", None), + name_to_id={}, + probe_store=InMemoryPermitProbeStore(), ): pass # yields without starting a task; kernel/observer untouched @@ -133,6 +174,7 @@ async def test_loop_records_observation_from_observer(db_pool: asyncpg.Pool) -> observer=observer, kernel=deps, name_to_id=name_to_id, + probe_store=InMemoryPermitProbeStore(), reconnect_delay_seconds=3600.0, # one pass, then idle (test cancels) ) ) @@ -154,11 +196,20 @@ async def test_loop_records_observation_from_observer(db_pool: asyncpg.Pool) -> @pytest.mark.unit -async def test_record_observation_bad_status_is_noop() -> None: +async def test_record_observation_bad_status_writes_probe_but_no_event() -> None: name = "hutch-bad" - # Unparseable status flattens to a no-op before any event-store access, - # so a None kernel is never touched. - await record_observation(cast("Kernel", None), _obs(name, "Garbage"), {name: uuid4()}) + probe_store = InMemoryPermitProbeStore() + # Unparseable status still writes the probe row (reach happened, the + # value just could not be classified), but never reaches the event + # store: EnclosurePermitStatus("Garbage") raises before kernel.event_store + # would ever be touched. `_ProbeOnlyKernel` exposes only `id_generator`, + # so an event-store access would AttributeError rather than pass silently. + await record_observation( + cast("Kernel", _ProbeOnlyKernel()), _obs(name, "Garbage"), {name: uuid4()}, probe_store + ) + rows = probe_store.all() + assert len(rows) == 1 + assert rows[0].status_claimed is True @pytest.mark.unit @@ -171,6 +222,7 @@ async def test_loop_logs_and_survives_record_failure() -> None: observer=observer, kernel=cast("Kernel", kernel), name_to_id={name: uuid4()}, + probe_store=InMemoryPermitProbeStore(), reconnect_delay_seconds=3600.0, ) ) @@ -192,6 +244,7 @@ async def test_loop_logs_and_survives_observer_iteration_failure() -> None: observer=observer, kernel=cast("Kernel", None), name_to_id={"hutch-iter-fail": uuid4()}, + probe_store=InMemoryPermitProbeStore(), reconnect_delay_seconds=3600.0, ) ) @@ -214,6 +267,7 @@ async def test_loop_cancellation_during_record_propagates() -> None: observer=observer, kernel=cast("Kernel", _CancelOnLoadKernel()), name_to_id={name: uuid4()}, + probe_store=InMemoryPermitProbeStore(), reconnect_delay_seconds=0.0, ) @@ -224,6 +278,7 @@ async def test_lifespan_nonempty_starts_and_cancels_monitor_task() -> None: observer=_FakeObserver([]), kernel=cast("Kernel", None), name_to_id={"hutch-lifespan": uuid4()}, + probe_store=InMemoryPermitProbeStore(), ): await asyncio.sleep(0) # let the background task start and park on reconnect # context exit cancels the task cleanly: no hang, no error surfaced @@ -238,8 +293,9 @@ async def test_run_monitor_startup_ready_waits_for_every_configured_code() -> No task = asyncio.create_task( run_enclosure_permit_monitor( observer=observer, - kernel=cast("Kernel", None), + kernel=cast("Kernel", _ProbeOnlyKernel()), name_to_id={code_a: uuid4(), code_b: uuid4()}, + probe_store=InMemoryPermitProbeStore(), reconnect_delay_seconds=3600.0, startup_ready=ready, ) @@ -256,21 +312,27 @@ async def test_run_monitor_startup_ready_waits_for_every_configured_code() -> No @pytest.mark.unit -async def test_run_monitor_startup_ready_set_when_pass_yields_nothing() -> None: - # An empty scope (observer.observe ends without yielding) still settles: - # startup_ready must not wait forever for a monitor that has nothing to say. +async def test_run_monitor_startup_ready_not_set_when_pass_yields_nothing() -> None: + # An empty scope (observer.observe ends without yielding anything) does + # NOT settle startup_ready: no code was ever heard from, so falsely + # reporting readiness here would let boot serve requests for an + # enclosure whose permit status this process never confirmed. The + # caller's own bounded wait (enclosure_permit_monitor_lifespan) is what + # eventually gives up on a deployment that cannot connect at all. ready = asyncio.Event() task = asyncio.create_task( run_enclosure_permit_monitor( observer=_FakeObserver([]), kernel=cast("Kernel", None), name_to_id={"hutch-empty": uuid4()}, + probe_store=InMemoryPermitProbeStore(), reconnect_delay_seconds=3600.0, startup_ready=ready, ) ) try: - await asyncio.wait_for(ready.wait(), timeout=2.0) + await asyncio.sleep(0.05) # let at least one empty pass complete + assert not ready.is_set() finally: task.cancel() with contextlib.suppress(asyncio.CancelledError): @@ -278,10 +340,12 @@ async def test_run_monitor_startup_ready_set_when_pass_yields_nothing() -> None: @pytest.mark.unit -async def test_run_monitor_startup_ready_set_on_iteration_failure() -> None: - # The observer raising outright (e.g. an immediate connect failure the - # observer itself could not turn into an Unknown) must still settle - # startup_ready, not hang it until the timeout on every reconnect pass. +async def test_run_monitor_startup_ready_not_set_on_total_iteration_failure() -> None: + # The observer raising outright, before ever yielding an observation, + # must NOT settle startup_ready either: no code was heard from, so + # settling here would be the same false-readiness bug as the + # empty-pass case above. The caller's bounded wait is what eventually + # gives up on this deployment. ready = asyncio.Event() observer = _BoomObserver() task = asyncio.create_task( @@ -289,12 +353,42 @@ async def test_run_monitor_startup_ready_set_on_iteration_failure() -> None: observer=observer, kernel=cast("Kernel", None), name_to_id={"hutch-iter-fail": uuid4()}, + probe_store=InMemoryPermitProbeStore(), reconnect_delay_seconds=3600.0, startup_ready=ready, ) ) try: - await asyncio.wait_for(ready.wait(), timeout=2.0) + await asyncio.wait_for(observer.observe_started.wait(), timeout=2.0) + await asyncio.sleep(0.05) + assert not ready.is_set() + finally: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + +@pytest.mark.unit +async def test_run_monitor_startup_ready_not_settled_by_probe_only_observation() -> None: + # F1 regression: a probe-only observation (observed_status=None, the + # shape a poll tick produces) must NOT settle startup readiness for its + # code. Settling here would let boot serve a stale permit_status this + # process never confirmed, exactly the window PR #642 closed. + name = "hutch-probe-only" + ready = asyncio.Event() + task = asyncio.create_task( + run_enclosure_permit_monitor( + observer=_FakeObserver([_obs(name, None)]), + kernel=cast("Kernel", _ProbeOnlyKernel()), + name_to_id={name: uuid4()}, + probe_store=InMemoryPermitProbeStore(), + reconnect_delay_seconds=3600.0, + startup_ready=ready, + ) + ) + try: + await asyncio.sleep(0.05) + assert not ready.is_set() finally: task.cancel() with contextlib.suppress(asyncio.CancelledError): @@ -309,8 +403,9 @@ async def test_lifespan_yields_once_ready_without_waiting_full_timeout() -> None name = "hutch-lifespan-ready" async with enclosure_permit_monitor_lifespan( observer=_FakeObserver([_obs(name, "Garbage")]), - kernel=cast("Kernel", None), + kernel=cast("Kernel", _ProbeOnlyKernel()), name_to_id={name: uuid4()}, + probe_store=InMemoryPermitProbeStore(), startup_timeout_seconds=3600.0, ): pass @@ -324,6 +419,7 @@ async def test_lifespan_gives_up_after_timeout_and_still_yields() -> None: observer=_HangingObserver(), kernel=cast("Kernel", None), name_to_id={"hutch-hang": uuid4()}, + probe_store=InMemoryPermitProbeStore(), startup_timeout_seconds=0.05, ): entered = True @@ -356,6 +452,7 @@ async def _raising_drain(*_args: object, **_kwargs: object) -> None: observer=_FakeObserver([_obs(name, "Garbage")]), kernel=cast("Kernel", _PoolOnlyKernel()), name_to_id={name: uuid4()}, + probe_store=InMemoryPermitProbeStore(), enclosure_projection_registry=ProjectionRegistry(), startup_timeout_seconds=5.0, ): @@ -375,6 +472,7 @@ async def test_lifespan_cancelled_during_startup_wait_still_cleans_up_task() -> observer=_HangingObserver(), kernel=cast("Kernel", None), name_to_id={"hutch-cancel-startup": uuid4()}, + probe_store=InMemoryPermitProbeStore(), startup_timeout_seconds=3600.0, # only our cancellation should end this ) entering = asyncio.ensure_future(cm.__aenter__()) @@ -408,6 +506,7 @@ async def test_lifespan_drains_projection_before_yielding(db_pool: asyncpg.Pool) observer=_FakeObserver([_obs(name, "Permitted")]), kernel=deps, name_to_id=name_to_id, + probe_store=InMemoryPermitProbeStore(), enclosure_projection_registry=registry, startup_timeout_seconds=5.0, ): @@ -419,6 +518,120 @@ async def test_lifespan_drains_projection_before_yielding(db_pool: asyncpg.Pool) assert row["permit_status"] == "Permitted" +# --- Permit probe trail ------------------------------------------------ + + +@pytest.mark.integration +async def test_record_observation_writes_a_probe_row_to_postgres(db_pool: asyncpg.Pool) -> None: + name = f"hutch-probe-pg-{uuid4().hex[:8]}" + deps = _deps_with(db_pool, permit_pvs={name: "pv"}) + name_to_id = await seed_enclosures(deps) + probe_store = PostgresPermitProbeStore(db_pool) + + await record_observation(deps, _obs(name, "Permitted"), name_to_id, probe_store) + + row = await db_pool.fetchrow( + "SELECT enclosure_id, source_kind, source_id, reach_tier, status_claimed " + "FROM entries_enclosure_permit_probes WHERE enclosure_id = $1", + name_to_id[name], + ) + assert row is not None + assert row["reach_tier"] == ReachTier.RELAYED.value + assert row["status_claimed"] is True + assert row["source_kind"] == "EpicsPv" + + +@pytest.mark.unit +async def test_probe_only_observation_writes_row_but_no_event() -> None: + # A probe-only observation (observed_status=None, the shape a poll tick + # produces) writes a probe row but never attempts a permit transition: + # `_ProbeOnlyKernel` exposes only `id_generator`, so any event-store + # access would AttributeError rather than silently succeed. + name = "hutch-probe-only-record" + probe_store = InMemoryPermitProbeStore() + await record_observation( + cast("Kernel", _ProbeOnlyKernel()), + _obs(name, None, reach_tier=ReachTier.UNREACHED), + {name: uuid4()}, + probe_store, + ) + rows = probe_store.all() + assert len(rows) == 1 + assert rows[0].status_claimed is False + assert rows[0].reach_tier is ReachTier.UNREACHED + + +@pytest.mark.unit +async def test_record_observation_skips_probe_write_when_schema_degraded() -> None: + # A degraded boot's event store is read-only (EventWritesDisabledError + # on append), so a probe row would assert reach for a process that + # cannot actually record what it observed. The correct signal is a + # gap in the trail, not a row. `_DegradedSchemaKernel` exposes ONLY + # `schema_posture`, so touching `id_generator` (part of building the + # probe row) would AttributeError if the skip fired too late. + name = "hutch-degraded" + probe_store = InMemoryPermitProbeStore() + with structlog.testing.capture_logs() as logs: + await record_observation( + cast("Kernel", _DegradedSchemaKernel()), + _obs(name, None), + {name: uuid4()}, + probe_store, + ) + assert probe_store.all() == [] + events = [e.get("event") for e in logs] + assert "enclosure_monitor.probe_skipped_degraded_schema" in events + + +@pytest.mark.integration +async def test_probe_store_failure_never_suppresses_the_permit_transition( + db_pool: asyncpg.Pool, +) -> None: + # The load-bearing lock (R6): a probe-store failure is a bookkeeping + # problem, not a safety problem, and must never take down the real + # transition. Driven against a real Postgres kernel so the assertion + # is "the transition actually landed", not merely "was attempted". + name = f"hutch-probe-fails-{uuid4().hex[:8]}" + deps = _deps_with(db_pool, permit_pvs={name: "pv"}) + name_to_id = await seed_enclosures(deps) + + with structlog.testing.capture_logs() as logs: + await record_observation( + deps, _obs(name, "Permitted"), name_to_id, _RaisingPermitProbeStore() + ) + + events = await _permit_events(deps, name_to_id[name]) + assert len(events) == 1 + assert events[0].payload["to_status"] == "Permitted" + log_events = [e.get("event") for e in logs] + assert "enclosure_monitor.probe_write_failed" in log_events + + +@pytest.mark.unit +async def test_run_monitor_writes_no_probe_rows_when_observer_yields_nothing() -> None: + # The anti-hook, stated in the module and design-lock docstrings: a + # probe row must come from evidence about the substrate, never from + # the loop's own liveness. An observer producing nothing, across + # several reconnect passes, must produce zero probe rows. + probe_store = InMemoryPermitProbeStore() + task = asyncio.create_task( + run_enclosure_permit_monitor( + observer=_FakeObserver([]), + kernel=cast("Kernel", None), + name_to_id={"hutch-inert": uuid4()}, + probe_store=probe_store, + reconnect_delay_seconds=0.01, + ) + ) + try: + await asyncio.sleep(0.1) # several reconnect passes at this cadence + assert probe_store.all() == [] + finally: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + class _FakeObserver: """Yields a fixed observation sequence once, then ends the stream.""" @@ -495,15 +708,54 @@ async def _drain(self) -> AsyncGenerator[EnclosureObservation]: class _PoolOnlyKernel: - """Kernel double exposing a non-None `pool` sentinel, nothing else. + """Kernel double exposing a non-None `pool` sentinel, plus a matched + `schema_posture` (the record_observation probe-write gate touches it + on every call) and `id_generator` (the observation this kernel is + driven with is status-bearing, so the probe write is attempted). Only used with a monkeypatched `drain_projections`, which never touches - `pool` for real; this just needs to satisfy the lifespan's + `pool` for real; `pool` just needs to satisfy the lifespan's `kernel.pool is not None` gate before reaching the drain call. """ def __init__(self) -> None: self.pool = object() + self.schema_posture = "matched" + self.id_generator = _FixedIdGenerator() + + +class _ProbeOnlyKernel: + """Kernel double exposing only `schema_posture` and `id_generator`. + + Used to prove a code path never reaches `kernel.event_store`: that + access would AttributeError here rather than silently succeeding. + """ + + def __init__(self) -> None: + self.schema_posture = "matched" + self.id_generator = _FixedIdGenerator() + + +class _DegradedSchemaKernel: + """Kernel double exposing only `schema_posture="degraded"`. + + Touching `id_generator` or `event_store` would mean the degraded-boot + probe-write skip fired too late (or not at all). + """ + + schema_posture = "degraded" + + +class _FixedIdGenerator: + def new_id(self) -> UUID: + return uuid4() + + +class _RaisingPermitProbeStore: + """`PermitProbeStore` double whose append always raises (R6 pin).""" + + async def append(self, rows: list[PermitProbe]) -> None: + raise RuntimeError("probe store boom") class _RaisingLoadKernel: @@ -512,6 +764,8 @@ class _RaisingLoadKernel: def __init__(self) -> None: self.load_attempted = asyncio.Event() self.event_store = self + self.schema_posture = "matched" + self.id_generator = _FixedIdGenerator() async def load(self, *, stream_type: str, stream_id: UUID) -> object: self.load_attempted.set() @@ -523,6 +777,8 @@ class _CancelOnLoadKernel: def __init__(self) -> None: self.event_store = self + self.schema_posture = "matched" + self.id_generator = _FixedIdGenerator() async def load(self, *, stream_type: str, stream_id: UUID) -> object: raise asyncio.CancelledError diff --git a/apps/api/tests/unit/api/test_enclosure_permit_observer.py b/apps/api/tests/unit/api/test_enclosure_permit_observer.py index 744d34455ae..1c1abbd2887 100644 --- a/apps/api/tests/unit/api/test_enclosure_permit_observer.py +++ b/apps/api/tests/unit/api/test_enclosure_permit_observer.py @@ -7,7 +7,8 @@ `ControlPort`. """ -from collections.abc import AsyncGenerator, AsyncIterator +import asyncio +from collections.abc import AsyncGenerator, AsyncIterator, Callable from datetime import UTC, datetime import pytest @@ -16,6 +17,7 @@ ControlPortEnclosureObserver, permit_status_from_reading, ) +from cora.enclosure.aggregates.enclosure import ReachTier from cora.enclosure.ports.enclosure_observer import ( EnclosureObservation, EnclosureObserverScope, @@ -128,8 +130,13 @@ class _ScriptedControlPort: """Fake `ControlPort`: replays a per-address reading script. Each address yields its scripted readings in order, then either ends - the stream cleanly or (when listed in `disconnect`) raises - `ControlNotConnectedError` to model a dropped subscription. + the stream cleanly, hangs forever (when listed in `hang`, modelling a + live ongoing subscription with no more traffic), or (when listed in + `disconnect`) raises `ControlNotConnectedError` to model a dropped + subscription. `read_results` scripts `read()` outcomes per address + for the poll path; a `Measurement` succeeds, an `Exception` instance + is raised, and an address with no results left raises + `ControlNotConnectedError`. """ def __init__( @@ -137,9 +144,13 @@ def __init__( *, readings: dict[str, list[Measurement]], disconnect: frozenset[str] = frozenset(), + hang: frozenset[str] = frozenset(), + read_results: dict[str, list[Measurement | Exception]] | None = None, ) -> None: self._readings = readings self._disconnect = disconnect + self._hang = hang + self._read_results = {k: list(v) for k, v in (read_results or {}).items()} def subscribe(self, address: str) -> AsyncIterator[Measurement]: return self._stream(address) @@ -147,16 +158,32 @@ def subscribe(self, address: str) -> AsyncIterator[Measurement]: async def _stream(self, address: str) -> AsyncGenerator[Measurement]: for reading in self._readings.get(address, []): yield reading + if address in self._hang: + await asyncio.Event().wait() # never released; models a live subscription + return # pragma: no cover - unreachable if address in self._disconnect: raise ControlNotConnectedError(address) + async def read(self, address: str) -> Measurement: + results = self._read_results.get(address) + if not results: + raise ControlNotConnectedError(address) + result = results.pop(0) + if isinstance(result, Exception): + raise result + return result + def _observer( - port: _ScriptedControlPort, permit_pvs: dict[str, str] + port: _ScriptedControlPort, + permit_pvs: dict[str, str], + *, + tick_seconds: float | None = None, ) -> ControlPortEnclosureObserver: return ControlPortEnclosureObserver( control_port=port, # type: ignore[arg-type] permit_pvs=permit_pvs, + tick_seconds=tick_seconds, ) @@ -266,3 +293,127 @@ async def test_observe_merges_multiple_pvs() -> None: ("hutch-b", "NotPermitted"), ("hutch-b", "Unknown"), } + + +async def _collect_until( + gen: AsyncGenerator[EnclosureObservation], + predicate: Callable[[list[EnclosureObservation]], bool], + *, + timeout_seconds: float = 2.0, +) -> list[EnclosureObservation]: + """Drain `gen` until `predicate(collected)` is true, then close it. + + Used for the poll tests below, whose generator never ends on its own + (a live push subscription hangs; the poller ticks forever), unlike + `_collect`, which relies on the scripted stream running out. + """ + collected: list[EnclosureObservation] = [] + + async def _drain() -> None: + async for observation in gen: + collected.append(observation) + if predicate(collected): + break + + try: + await asyncio.wait_for(_drain(), timeout=timeout_seconds) + finally: + await gen.aclose() + return collected + + +@pytest.mark.unit +async def test_poll_disabled_by_default_emits_nothing_extra() -> None: + # tick_seconds defaults to None: no poll task is created at all, so a + # live (never-ending) subscription with no push traffic yields nothing. + port = _ScriptedControlPort(readings={"pvA": []}, hang=frozenset({"pvA"})) + observer = _observer(port, {"hutch-a": "pvA"}) + gen = observer.observe(EnclosureObserverScope(enclosure_codes=frozenset({"hutch-a"}))) + try: + with pytest.raises(TimeoutError): + await asyncio.wait_for(anext(gen), timeout=0.05) + finally: + await gen.aclose() + + +@pytest.mark.unit +async def test_poll_emits_relayed_probe_on_successful_read() -> None: + port = _ScriptedControlPort( + readings={"pvA": []}, + hang=frozenset({"pvA"}), # push subscription stays open, no traffic + read_results={"pvA": [_reading(1)]}, + ) + observer = _observer(port, {"hutch-a": "pvA"}, tick_seconds=0.01) + gen = observer.observe(EnclosureObserverScope(enclosure_codes=frozenset({"hutch-a"}))) + + collected = await _collect_until(gen, lambda obs: len(obs) >= 1) + + assert len(collected) == 1 + probe = collected[0] + assert probe.enclosure_code == "hutch-a" + assert probe.observed_status is None # probe-only: makes no status claim + assert probe.reach_tier is ReachTier.RELAYED + assert probe.source_kind == "EpicsPv" + assert probe.source_id == "pvA" + + +@pytest.mark.unit +async def test_poll_emits_unreached_probe_on_failed_read() -> None: + port = _ScriptedControlPort( + readings={"pvA": []}, + hang=frozenset({"pvA"}), + read_results={"pvA": [ControlNotConnectedError("pvA")]}, + ) + observer = _observer(port, {"hutch-a": "pvA"}, tick_seconds=0.01) + gen = observer.observe(EnclosureObserverScope(enclosure_codes=frozenset({"hutch-a"}))) + + collected = await _collect_until(gen, lambda obs: len(obs) >= 1) + + assert len(collected) == 1 + probe = collected[0] + assert probe.observed_status is None + assert probe.reach_tier is ReachTier.UNREACHED + + +@pytest.mark.unit +async def test_poll_survives_a_disconnected_sibling_pump() -> None: + """A poller keeps ticking for its own PV even after ANOTHER PV's pump + disconnects. + + `_drain` only returns once EVERY pump has finished (pvB's subscribe + hangs forever here, modelling a live PV with nothing to say), so + pvA's poller keeps running and ticking after pvA's pump has already + emitted its disconnect Unknown and exited. This is the load-bearing + property that makes the poller a sibling of its pump, not a stage + nested inside it: nesting would kill the poller the moment its own + pump dies, with no way left to notice that PV's recovery. + """ + port = _ScriptedControlPort( + readings={"pvB": []}, + hang=frozenset({"pvB"}), # pvB's pump never finishes + disconnect=frozenset({"pvA"}), # pvA's pump dies immediately + read_results={"pvA": [_reading(1), _reading(1)]}, + ) + observer = _observer(port, {"hutch-a": "pvA", "hutch-b": "pvB"}, tick_seconds=0.01) + gen = observer.observe( + EnclosureObserverScope(enclosure_codes=frozenset({"hutch-a", "hutch-b"})) + ) + + def _seen_disconnect_and_a_probe(obs: list[EnclosureObservation]) -> bool: + disconnected = any( + o.enclosure_code == "hutch-a" and o.observed_status == "Unknown" for o in obs + ) + probed = any(o.enclosure_code == "hutch-a" and o.observed_status is None for o in obs) + return disconnected and probed + + collected = await _collect_until(gen, _seen_disconnect_and_a_probe, timeout_seconds=2.0) + + disconnect_obs = [ + o for o in collected if o.enclosure_code == "hutch-a" and o.observed_status == "Unknown" + ] + probe_obs = [ + o for o in collected if o.enclosure_code == "hutch-a" and o.observed_status is None + ] + assert disconnect_obs, "pump A's disconnect Unknown must still be observed" + assert probe_obs, "the poller for A must keep ticking after A's pump has died" + assert probe_obs[0].reach_tier is ReachTier.RELAYED diff --git a/apps/api/tests/unit/enclosure/test_enclosure_observer.py b/apps/api/tests/unit/enclosure/test_enclosure_observer.py index bcebc0312ed..d71ce1ccfc4 100644 --- a/apps/api/tests/unit/enclosure/test_enclosure_observer.py +++ b/apps/api/tests/unit/enclosure/test_enclosure_observer.py @@ -20,7 +20,7 @@ import pytest -from cora.enclosure.aggregates.enclosure import EnclosurePermitStatus +from cora.enclosure.aggregates.enclosure import EnclosurePermitStatus, ReachTier from cora.enclosure.ports.enclosure_observer import ( AlwaysPermittedEnclosureObserver, EnclosureObservation, @@ -124,11 +124,25 @@ async def test_always_permitted_observer_yields_observations_for_each_distinct_c assert sorted(codes_seen) == sorted({_HUTCH_A, _HUTCH_B}) +@pytest.mark.unit +async def test_always_permitted_observer_emits_relayed_reach_tier() -> None: + """The stub always delivers a status claim, the same shape as a real + push delivery, so it is RELAYED, not UNREACHED: pairing a delivered + status with a no-reach-evidence tier is the illegal combination the + port docstring warns adapters against.""" + observer = AlwaysPermittedEnclosureObserver() + scope = EnclosureObserverScope(enclosure_codes=frozenset({_HUTCH_A})) + observations = [obs async for obs in observer.observe(scope)] + assert len(observations) == 1 + assert observations[0].reach_tier is ReachTier.RELAYED + + @pytest.mark.unit def test_enclosure_observation_is_frozen_dataclass() -> None: obs = EnclosureObservation( enclosure_code=_HUTCH_A, observed_status=EnclosurePermitStatus.PERMITTED.value, + reach_tier=ReachTier.RELAYED, observed_at=_EPOCH, source_kind="Stub", source_id="always-permitted", diff --git a/docs/architecture/modules/enclosure/index.md b/docs/architecture/modules/enclosure/index.md index 1db534718f5..4781805826c 100644 --- a/docs/architecture/modules/enclosure/index.md +++ b/docs/architecture/modules/enclosure/index.md @@ -153,6 +153,45 @@ CREATE INDEX proj_enclosure_summary_gate_idx The address tuple `(facility_code, name)` is enforced unique at the projection because aggregates cannot enforce cross-stream invariants. The UNIQUE INDEX is partial on `WHERE lifecycle = 'Active'` so a decommissioned Enclosure does not hold the address against re-registration. CHECK constraints lock all enum values day-one so a future widening of the operational axis lands without a constraint migration. No FK constraints to other projections: rebuild order is arbitrary, and cross-projection consistency is event-driven. The `last_*` columns live only on the projection (the slim-aggregate rule keeps the aggregate state cheap to fold); the `last_source_kind` / `last_source_id` split mirrors the colon-delimited wire form so the gate-side queries can filter on adapter kind without LIKE patterns. The table carries two clocks and they answer different questions: `last_permit_status_changed_at` is CORA's ingest time for the last permit-status change, from the event's `occurred_at`, while `last_source_observed_at` is the substrate's own time for the reading behind it and is NULL whenever the substrate reported none. Both advance only on a change, because the decider emits nothing for an identical-status observation, so a stale value means "no transition since" rather than "not observed since". +A stale value means "no transition since", never "not observed since" -- and `proj_enclosure_summary` alone cannot distinguish the two, which is why the permit probe trail exists. + +`entries_enclosure_permit_probes`: + +```sql title="entries_enclosure_permit_probes" +CREATE TABLE entries_enclosure_permit_probes ( + event_id UUID PRIMARY KEY, + enclosure_id UUID NOT NULL, + source_kind TEXT NOT NULL CHECK (length(source_kind) BETWEEN 1 AND 50), + source_id TEXT NOT NULL CHECK (length(source_id) BETWEEN 1 AND 200), + reach_tier TEXT NOT NULL CHECK (length(reach_tier) BETWEEN 1 AND 32), + status_claimed BOOLEAN NOT NULL, + recorded_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX entries_enclosure_permit_probes_enclosure_recorded_idx + ON entries_enclosure_permit_probes (enclosure_id, recorded_at DESC); +``` + +Append-only (INSERT-only role grant, `UPDATE` / `DELETE` / `TRUNCATE` REVOKEd): one row per observation the permit monitor's `EnclosureObserver` surfaces, whether or not it caused a `permit_status` transition. This is the record of whether CORA could reach the substrate, kept deliberately separate from `EnclosurePermitObserved`'s record of what the interlock said; the row never carries the observed permit value itself, so it cannot become a second source of truth for permit status. `reach_tier` is `RELAYED` (CORA received or fetched a value through the configured channel) or `UNREACHED` (it could not, this tick); a stronger tier for a confirmed direct round trip to the authoritative source, as opposed to an intermediary such as an EPICS CA gateway that may answer from its own cache, is deliberately not defined until a deployment can demonstrate one. `status_claimed` distinguishes a push delivery or a real substrate disconnect (both carry a permit-status claim) from a periodic re-affirmation poll (which intentionally makes none); it is a fact about the probe, not the hutch. `recorded_at` (DB `DEFAULT now()`) is the only clock on this row: no producer-asserted timestamp crosses the `EnclosureObserver` port, so consumer-side queueing lag can forward-date a row relative to when reach actually happened, with nothing on the row to detect it in v1. + +Reading the trail (no query slice exists yet; this is a direct read against the table, same posture as the `proj_enclosure_summary` psql query above): + +```sql title="latest confirmed reach per enclosure" +SELECT e.enclosure_id, e.name, p.reach_tier, p.recorded_at, + now() - p.recorded_at AS since_last_probe +FROM proj_enclosure_summary e +LEFT JOIN LATERAL ( + SELECT reach_tier, recorded_at + FROM entries_enclosure_permit_probes + WHERE enclosure_id = e.enclosure_id + ORDER BY recorded_at DESC + LIMIT 1 +) p ON true +WHERE e.lifecycle = 'Active'; +``` + +One caveat worth stating plainly rather than discovering at 3am: a `RELAYED` row proves CORA reached the configured channel, never that the underlying substrate value is current. At a deployment reading through a caching intermediary, a `RELAYED` trail can stay dense through an upstream outage the intermediary is masking. This is not visible from `reach_tier` alone; it requires knowing the deployment's topology, not just querying this table. A related failure mode is closed rather than merely documented: no probe row is written at all while a process runs degraded under the schema-version mismatch override (whose event store is read-only), so a gap in the trail, not a misleadingly dense one, is what a reader sees across that window. + ## Cross-Module boundaries | Module | Relationship | What's exchanged | diff --git a/infra/atlas/migrations/20260810000000_init_entries_enclosure_permit_probes.sql b/infra/atlas/migrations/20260810000000_init_entries_enclosure_permit_probes.sql new file mode 100644 index 00000000000..f8ca3c0dd56 --- /dev/null +++ b/infra/atlas/migrations/20260810000000_init_entries_enclosure_permit_probes.sql @@ -0,0 +1,83 @@ +-- Permit probe trail: append-only record of CORA's reach to the +-- enclosure permit substrate, separate from the EnclosurePermitObserved +-- transition events. +-- +-- See [[project_enclosure_permit_probe_design]]. The permit-status +-- projection advances only on a CHANGE (the decider is status-change- +-- only), so a stale value means "no transition since", never "not +-- observed since". This table is the mechanism that tells the two +-- apart: one row per observation the monitor's EnclosureObserver +-- surfaces, whether or not it caused a transition. +-- +-- ## Two facts, two homes +-- +-- `proj_enclosure_summary.permit_status` answers what the interlock +-- said; this table answers whether CORA could reach it. Neither +-- substitutes for the other. This table does NOT carry the observed +-- permit value, so it cannot become a second source of truth for +-- permit status. +-- +-- ## reach_tier: two values shipped, a third reserved +-- +-- 'RELAYED' means CORA received or fetched a value through the +-- configured channel; 'UNREACHED' means it could not, this tick. +-- Reach is not the same as belief: a Bad-quality reading is a RELAYED +-- probe with an unbelievable value, because reachability and +-- believability are different questions (permit_status answers the +-- second). A stronger tier for a confirmed direct round trip to the +-- authoritative source is deliberately NOT defined yet: no producer in +-- this codebase can currently prove one (2-BM reads through a caching +-- EPICS CA gateway), and shipping an unearned strong claim is worse +-- than shipping none. Adding a value later needs no migration, since +-- the column is a length-CHECK, not a value-enumerating CHECK. +-- +-- `status_claimed` distinguishes, within a single reach_tier value, +-- whether the observation also carried a permit-status claim (a push +-- delivery, or a real substrate disconnect) versus being probe-only (a +-- periodic re-affirmation read that intentionally makes no status +-- claim). This is a fact about the PROBE, not the hutch, so recording +-- it does not create a second source of truth for permit status. +-- +-- ## Append-only INSERT, not UPSERT +-- +-- One row per observation. An UPSERT would need UPDATE, which the +-- append-only cora_app role is REVOKEd from (test_migration_revokes +-- enforces this for every entries_* table). Mirrors +-- entries_run_feed_heartbeats and the other entries_* tables. +-- +-- ## recorded_at is the only anchor +-- +-- No producer-asserted timestamp crosses the EnclosureObserver port +-- (see the port docstring), so this table carries only the DB write +-- time. Consumer-side queueing lag can forward-date a row relative to +-- when reach actually happened; there is no second timestamp to detect +-- this in v1. + +CREATE TABLE entries_enclosure_permit_probes ( + event_id uuid PRIMARY KEY, + enclosure_id uuid NOT NULL, + source_kind text NOT NULL CHECK (length(source_kind) BETWEEN 1 AND 50), + source_id text NOT NULL CHECK (length(source_id) BETWEEN 1 AND 200), + reach_tier text NOT NULL CHECK (length(reach_tier) BETWEEN 1 AND 32), + status_claimed boolean NOT NULL, + recorded_at timestamptz NOT NULL DEFAULT now() +); + +COMMENT ON COLUMN entries_enclosure_permit_probes.reach_tier IS + 'RELAYED or UNREACHED in v1. A stronger direct-round-trip tier is reserved and unused; see the migration header.'; + +-- Supports the coverage read: latest probe per enclosure via +-- MAX(recorded_at) / ORDER BY recorded_at DESC LIMIT 1. +CREATE INDEX entries_enclosure_permit_probes_enclosure_recorded_idx + ON entries_enclosure_permit_probes (enclosure_id, recorded_at DESC); + +-- cora_app has no table-level default privileges in this database (only +-- ALTER DEFAULT PRIVILEGES ... ON SEQUENCES exists); grant explicitly +-- rather than repeating the false "inherits via ALTER DEFAULT +-- PRIVILEGES" claim several other entries_* migration headers carry. +GRANT SELECT, INSERT ON entries_enclosure_permit_probes TO cora_app; + +-- Append-only at the role level (project_immutability_guarantee.md): +-- this removes mutation privileges so the append-only shape cannot be +-- silently broken. +REVOKE UPDATE, DELETE, TRUNCATE ON entries_enclosure_permit_probes FROM cora_app; -- atlas:safety:allow=append-only revoke, not destructive DDL diff --git a/infra/atlas/migrations/atlas.sum b/infra/atlas/migrations/atlas.sum index 060c975e672..febbb34e5a4 100644 --- a/infra/atlas/migrations/atlas.sum +++ b/infra/atlas/migrations/atlas.sum @@ -1,4 +1,4 @@ -h1:LiFfsQkvQmASecd/QFx9VE1SOE0RHKlpWw++Tqv5XZo= +h1:bdALzYDPVBnS+sQ6HlFKoV4ZwWy02BiLdJGYVeBbgGg= 20260509120000_init_events.sql h1:GmgCZKfaqXu1m96/cKAks2vhaLWTdEaHTLkFtUo9FXg= 20260509170000_init_idempotency.sql h1:Nbu8DIE4Sv1WiHw3G22+tYffPhKc5Jryw3PMK8wB2zY= 20260510010000_add_event_id.sql h1:RbtYP6uMnOB20zhJ9dNXUi4YVqbmlEzf562pmygnRW8= @@ -161,3 +161,4 @@ h1:LiFfsQkvQmASecd/QFx9VE1SOE0RHKlpWw++Tqv5XZo= 20260713000000_init_proj_budget_allocation_summary.sql h1:72j+IH7DwQ9xmV1srtNszWNLhBmIu3nDljwIHYYnm8E= 20260729120000_add_proj_data_dataset_summary_checksum.sql h1:Zk3BmIBMjck0Q4OkKzP1wU93NLWC2b5IRV0iM0ihlX0= 20260809190000_split_enclosure_permit_transition_and_source_times.sql h1:xjTv+1Eq6Ujg40E4y0mOK5nIlT8pXKhdp3ve7No+uvQ= +20260810000000_init_entries_enclosure_permit_probes.sql h1:jSvy8jXhHwZbq1A4AMqKEL3uviaJtlyGLCGvISfevd4=